Java Stream 源碼分析

時間 2020-12-03

標籤 java git github 數據庫數組安全服務器 app 框架 less 欄目 Java 简体版

原文原文鏈接

前言

Java 8 的 Stream 使得代碼更加簡潔易懂，本篇文章深刻分析 Java Stream 的工做原理，並探討 Steam 的性能問題。java

Java 8 集合中的 Stream 至關於高級版的 Iterator，它能夠經過 Lambda 表達式對集合進行各類很是便利、高效的聚合操做（Aggregate Operation），或者大批量數據操做 (Bulk Data Operation)。git

Stream的聚合操做與數據庫SQL的聚合操做sorted、filter、map等相似。咱們在應用層就能夠高效地實現相似數據庫SQL的聚合操做了，而在數據操做方面，Stream不只能夠經過串行的方式實現數據操做，還能夠經過並行的方式處理大批量數據，提升數據的處理效率。github

操做分類

官方將 Stream 中的操做分爲兩大類：數據庫

中間操做（Intermediate operations），只對操做進行了記錄，即只會返回一個流，不會進行計算操做。
終結操做（Terminal operations），實現了計算操做。

中間操做又能夠分爲：數組

無狀態（Stateless）操做，元素的處理不受以前元素的影響。
有狀態（Stateful）操做，指該操做只有拿到全部元素以後才能繼續下去。

終結操做又能夠分爲：安全

短路（Short-circuiting）操做，指遇到某些符合條件的元素就能夠獲得最終結果
非短路（Unshort-circuiting）操做，指必須處理完全部元素才能獲得最終結果。

操做分類詳情以下圖所示：服務器

源碼結構

Stream 相關類和接口的繼承關係以下圖所示：app

BaseStream

最頂端的接口類，定義了流的基本接口方法，最主要的方法爲 spliterator、isParallel。框架

Stream

最頂端的接口類。定義了流的經常使用方法，例如 map、filter、sorted、limit、skip、collect 等。less

ReferencePipeline

ReferencePipeline 是一個結構類，定義內部類組裝了各類操做流，定義了Head、StatelessOp、StatefulOp三個內部類，實現了 BaseStream 與 Stream 的接口方法。

Sink

Sink 接口定義了 Stream 之間的操做行爲，包含 begin()、end()、cancellationRequested()、accpt()四個方法。ReferencePipeline最終會將整個 Stream 流操做組裝成一個調用鏈，而這條調用鏈上的各個 Stream 操做的上下關係就是經過 Sink 接口協議來定義實現的。

操做疊加

Stream 的基礎用法就再也不敘述了，這裏從一段代碼開始，分析 Stream 的工做原理。

@Test
public void testStream() {
    List<String> names = Arrays.asList("kotlin", "java", "go");
    int maxLength = names.stream().filter(name -> name.length() <= 4).map(String::length)
            .max(Comparator.naturalOrder()).orElse(-1);
    System.out.println(maxLength);
}

當使用 Stream 時，主要有 3 部分組成，下面一一講解。

加載數據源

調用 names.stream() 方法，會初次加載 ReferencePipeline 的 Head 對象，此時爲加載數據源操做。

java.util.Collection#stream

default Stream<E> stream() {
    return StreamSupport.stream(spliterator(), false);
}

StreamSupport 類中的 stream 方法，初始化了一個 ReferencePipeline的 Head 內部類對象。

java.util.stream.StreamSupport#stream(java.util.Spliterator , boolean)

public static <T> Stream<T> stream(Spliterator<T> spliterator, boolean parallel) {
    Objects.requireNonNull(spliterator);
    return new ReferencePipeline.Head<>(spliterator,
                                        StreamOpFlag.fromCharacteristics(spliterator),
                                        parallel);
}

中間操做

接着爲 filter(name -> name.length() <= 4).mapToInt(String::length)，是中間操做，分爲無狀態中間操做 StatelessOp 對象和有狀態操做 StatefulOp 對象，此時的 Stage 並無執行，而是經過AbstractPipeline 生成了一箇中間操做 Stage 鏈表。

java.util.stream.ReferencePipeline#filter

@Override
public final Stream<P_OUT> filter(Predicate<? super P_OUT> predicate) {
    Objects.requireNonNull(predicate);
    return new StatelessOp<P_OUT, P_OUT>(this, StreamShape.REFERENCE,
                                    StreamOpFlag.NOT_SIZED) {
        @Override
        Sink<P_OUT> opWrapSink(int flags, Sink<P_OUT> sink) {
            return new Sink.ChainedReference<P_OUT, P_OUT>(sink) {
                @Override
                public void begin(long size) {
                    downstream.begin(-1);
                }

                @Override
                public void accept(P_OUT u) {
                    if (predicate.test(u))
                        downstream.accept(u);
                }
            };
        }
    };
}

java.util.stream.ReferencePipeline#map

@Override
@SuppressWarnings("unchecked")
public final <R> Stream<R> map(Function<? super P_OUT, ? extends R> mapper) {
    Objects.requireNonNull(mapper);
    return new StatelessOp<P_OUT, R>(this, StreamShape.REFERENCE,
                                    StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
        @Override
        Sink<P_OUT> opWrapSink(int flags, Sink<R> sink) {
            return new Sink.ChainedReference<P_OUT, R>(sink) {
                @Override
                public void accept(P_OUT u) {
                    downstream.accept(mapper.apply(u));
                }
            };
        }
    };
}

能夠看到 filter 和 map 方法都返回了一個新的 StatelessOp 對象。new StatelessOp 將會調用父類 AbstractPipeline 的構造函數，這個構造函數將先後的 Stage 聯繫起來，生成一個 Stage 鏈表：

AbstractPipeline(AbstractPipeline<?, E_IN, ?> previousStage, int opFlags) {
    if (previousStage.linkedOrConsumed)
        throw new IllegalStateException(MSG_STREAM_LINKED);
    previousStage.linkedOrConsumed = true;
    previousStage.nextStage = this;

    this.previousStage = previousStage;
    this.sourceOrOpFlags = opFlags & StreamOpFlag.OP_MASK;
    this.combinedFlags = StreamOpFlag.combineOpFlags(opFlags, previousStage.combinedFlags);
    this.sourceStage = previousStage.sourceStage;
    if (opIsStateful())
        sourceStage.sourceAnyStateful = true;
    this.depth = previousStage.depth + 1;
}

終結操做

最後爲 max(Comparator.naturalOrder())，是終結操做，會生成一個最終的 Stage，經過這個 Stage 觸發以前的中間操做，從最後一個Stage開始，遞歸產生一個Sink鏈。

java.util.stream.ReferencePipeline#max

@Override
public final Optional<P_OUT> max(Comparator<? super P_OUT> comparator) {
    return reduce(BinaryOperator.maxBy(comparator));
}

最終調用到 java.util.stream.AbstractPipeline#wrapSink，這個方法會調用 opWrapSink 生成一個 Sink 鏈表，對應到本文的例子，就是 filter 和 map 操做。

@Override
@SuppressWarnings("unchecked")
final <P_IN> Sink<P_IN> wrapSink(Sink<E_OUT> sink) {
    Objects.requireNonNull(sink);

    for ( @SuppressWarnings("rawtypes") AbstractPipeline p=AbstractPipeline.this; p.depth > 0; p=p.previousStage) {
        sink = p.opWrapSink(p.previousStage.combinedFlags, sink);
    }
    return (Sink<P_IN>) sink;
}

在上面 opWrapSink 上斷點調試，發現最終會調用到本例中的 filter 和 map 操做。

wrapAndCopyInto 生成 Sink 鏈表後，會經過 copyInfo 方法執行 Sink 鏈表的具體操做。

@Override
final <P_IN> void copyInto(Sink<P_IN> wrappedSink, Spliterator<P_IN> spliterator) {
    Objects.requireNonNull(wrappedSink);

    if (!StreamOpFlag.SHORT_CIRCUIT.isKnown(getStreamAndOpFlags())) {
        wrappedSink.begin(spliterator.getExactSizeIfKnown());
        spliterator.forEachRemaining(wrappedSink);
        wrappedSink.end();
    }
    else {
        copyIntoWithCancel(wrappedSink, spliterator);
    }
}

上面的核心代碼是：

spliterator.forEachRemaining(wrappedSink);

java.util.Spliterators.ArraySpliterator#forEachRemaining

@Override
public void forEachRemaining(Consumer<? super T> action) {
    Object[] a; int i, hi; // hoist accesses and checks from loop
    if (action == null)
        throw new NullPointerException();
    if ((a = array).length >= (hi = fence) &&
        (i = index) >= 0 && i < (index = hi)) {
        do { action.accept((T)a[i]); } while (++i < hi);
    }
}

斷點調試，能夠發現首先進入了 filter 的 Sink，其中 accept 方法的入參是 list 中的第一個元素「kotlin」（代碼中的 3 個元素是："kotlin", "java", "go"）。filter 的傳入是一個 Lambda 表達式：

filter(name -> name.length() <= 4)

顯然這個第一個元素「kotlin」的 predicate 是不會進入的。

對於第二個元素「java」，predicate.test 會返回 true（字符串「java」的長度<=4），則會進入 map 的 accept 方法。

本次調用 accept 方法時，empty 爲 false，會將 map 後的結果（int 類型的 4）賦值給 t。

public static <T> TerminalOp<T, Optional<T>>
makeRef(BinaryOperator<T> operator) {
    Objects.requireNonNull(operator);
    class ReducingSink
            implements AccumulatingSink<T, Optional<T>, ReducingSink> {
        private boolean empty;
        private T state;

        public void begin(long size) {
            empty = true;
            state = null;
        }

        @Override
        public void accept(T t) {
            if (empty) {
                empty = false;
                state = t;
            } else {
                state = operator.apply(state, t);
            }
        }

        ……
        }
}

對於第三個元素「go」，也會進入 accept 方法，此時 empty 爲 true, map 後的結果（int 類型的 2）會與上次的結果 4 經過自定義的比較器相比較，存入符合結果的值。

public static <T> BinaryOperator<T> maxBy(Comparator<? super T> comparator) {
    Objects.requireNonNull(comparator);
    return (a, b) -> comparator.compare(a, b) >= 0 ? a : b;
}

本文代碼中的 max 傳入的比較器爲：

max(Comparator.naturalOrder())

至此會返回 int 類型的 4。

並行處理

上面的例子是串行處理的，若是要改爲並行也很簡單，只須要在 stream() 方法後加上 parallel() 就能夠了，並行代碼能夠寫成：

@Test
public void testStream() {
    List<String> names = Arrays.asList("kotlin", "java", "go");
    int maxLength = names.stream().parallel().filter(name -> name.length() <= 4)
            .map(String::length).max(Comparator.naturalOrder()).orElse(-1);
    System.out.println(maxLength);
}

Stream 的並行處理在執行終結操做以前，跟串行處理的實現是同樣的。而在調用終結方法以後，實現的方式就有點不太同樣，會調用 TerminalOp 的 evaluateParallel 方法進行並行處理。

final <R> R evaluate(TerminalOp<E_OUT, R> terminalOp) {
    assert getOutputShape() == terminalOp.inputShape();
    if (linkedOrConsumed)
        throw new IllegalStateException(MSG_STREAM_LINKED);
    linkedOrConsumed = true;

    return isParallel()
            ? terminalOp.evaluateParallel(this, sourceSpliterator(terminalOp.getOpFlags()))
            : terminalOp.evaluateSequential(this, sourceSpliterator(terminalOp.getOpFlags()));
}

核心是使用了 ForkJoin 框架，對 Stream 處理進行分片，最終會調用下面的代碼，這裏就不展開分析了。

java.util.stream.AbstractTask#compute

@Override
public void compute() {
    Spliterator<P_IN> rs = spliterator, ls; // right, left spliterators
    long sizeEstimate = rs.estimateSize();
    long sizeThreshold = getTargetSize(sizeEstimate);
    boolean forkRight = false;
    @SuppressWarnings("unchecked") K task = (K) this;
    while (sizeEstimate > sizeThreshold && (ls = rs.trySplit()) != null) {
        K leftChild, rightChild, taskToFork;
        task.leftChild  = leftChild = task.makeChild(ls);
        task.rightChild = rightChild = task.makeChild(rs);
        task.setPendingCount(1);
        if (forkRight) {
            forkRight = false;
            rs = ls;
            task = leftChild;
            taskToFork = rightChild;
        }
        else {
            forkRight = true;
            task = rightChild;
            taskToFork = leftChild;
        }
        taskToFork.fork();
        sizeEstimate = rs.estimateSize();
    }
    task.setLocalResult(task.doLeaf());
    task.tryComplete();
}

並行錯誤的使用方法

@Test
public void testParallelWrong() {
    List<Integer> parallelList = new ArrayList<>();
    IntStream.range(0, 1000).boxed().parallel().filter(i -> i % 2 == 1)
            .forEach(parallelList::add);
    System.out.println(parallelList.size());
}

上面的輸出結果會常常小於500，這是由於 parallelList 的類型是 ArrayList，並非線程安全的，在執行 add 操做時，可能正好遇上擴容或者線程被佔用，會覆蓋其餘線程的賦好的值。

並行正確的使用方法

@Test
public void testParallelRight() {
    List<Integer> parallelList = IntStream.range(0, 1000).boxed().parallel()
            .filter(i -> i % 2 == 1).collect(Collectors.toList());
    System.out.println(parallelList.size());
}

性能

下面的文章參考自：JavaLambdaInternals/8-Stream Performance.md，侵刪。

爲保證測試結果然實可信，咱們將JVM運行在-server模式下，測試數據在GB量級，測試機器採用常見的商用服務器，配置以下：

OS	CentOS 6.7 x86_64
CPU	Intel Xeon X5675, 12M Cache 3.06 GHz, 6 Cores 12 Threads
內存	96GB
JDK	java version 1.8.0_91, Java HotSpot(TM) 64-Bit Server VM

測試所用代碼在這裏，測試結果彙總.

測試方法和測試數據

性能測試並非容易的事，Java性能測試更費勁，由於虛擬機對性能的影響很大，JVM對性能的影響有兩方面：

GC的影響。GC的行爲是Java中很很差控制的一塊，爲增長肯定性，咱們手動指定使用CMS收集器，並使用10GB固定大小的堆內存。具體到JVM參數就是-XX:+UseConcMarkSweepGC -Xms10G -Xmx10G
JIT(Just-In-Time)即時編譯技術。即時編譯技術會將熱點代碼在JVM運行的過程當中編譯成本地代碼，測試時咱們會先對程序預熱，觸發對測試函數的即時編譯。相關的JVM參數是-XX:CompileThreshold=10000。

Stream並行執行時用到ForkJoinPool.commonPool()獲得的線程池，爲控制並行度咱們使用Linux的taskset命令指定JVM可用的核數。

測試數據由程序隨機生成。爲防止一次測試帶來的抖動，測試4次求出平均時間做爲運行時間。