前言

Java 8 的 Stream 使得代碼更加簡潔易懂，本篇文章深入分析 Java Stream 的作業原理，并探討 Steam 的性能問題，

Java 8 集合中的 Stream 相當于高級版的 Iterator，它可以通過 Lambda 運算式對集合進行各種非常便利、高效的聚合操作（Aggregate Operation），或者大批量資料操作 (Bulk Data Operation)，

Stream的聚合操作與資料庫SQL的聚合操作sorted、filter、map等類似，我們在應用層就可以高效地實作類似資料庫SQL的聚合操作了，而在資料操作方面，Stream不僅可以通過串行的方式實作資料操作，還可以通過并行的方式處理大批量資料，提高資料的處理效率，

操作分類

官方將 Stream 中的操作分為兩大類：

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

中間操作又可以分為：

• 無狀態（Stateless）操作，元素的處理不受之前元素的影響，
• 有狀態（Stateful）操作，指該操作只有拿到所有元素之后才能繼續下去，

終結操作又可以分為：

• 短路（Short-circuiting）操作，指遇到某些符合條件的元素就可以得到最終結果
• 非短路（Unshort-circuiting）操作，指必須處理完所有元素才能得到最終結果，

操作分類詳情如下圖所示：

原始碼結構

Stream 相關類和介面的繼承關系如下圖所示：

BaseStream

最頂端的介面類，定義了流的基本介面方法，最主要的方法為 spliterator、isParallel，

Stream

最頂端的介面類，定義了流的常用方法，例如 map、filter、sorted、limit、skip、collect 等，

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量級，測驗機器采用常見的商用服務器，配置如下：

測驗所用代碼在這里，測驗結果匯總.

測驗方法和測驗資料

性能測驗并不是容易的事，Java性能測驗更費勁，因為虛擬機對性能的影響很大，JVM對性能的影響有兩方面：

1. GC的影響，GC的行為是Java中很不好控制的一塊，為增加確定性，我們手動指定使用CMS收集器，并使用10GB固定大小的堆記憶體，具體到JVM引數就是-XX:+UseConcMarkSweepGC -Xms10G -Xmx10G
2. JIT(Just-In-Time)即時編譯技術，即時編譯技識訓將熱點代碼在JVM運行的程序中編譯成本地代碼，測驗時我們會先對程式預熱，觸發對測驗函式的即時編譯，相關的JVM引數是-XX:CompileThreshold=10000，

Stream并行執行時用到ForkJoinPool.commonPool()得到的執行緒池，為控制并行度我們使用Linux的taskset命令指定JVM可用的核數，

測驗資料由程式隨機生成，為防止一次測驗帶來的抖動，測驗4次求出平均時間作為運行時間，

實驗一 基本型別迭代

測驗內容：找出整型陣列中的最小值，對比for回圈外部迭代和Stream API內部迭代性能，

測驗程式IntTest，測驗結果如下圖：

圖中展示的是for回圈外部迭代耗時為基準的時間比值，分析如下：

1. 對于基本型別Stream串行迭代的性能開銷明顯高于外部迭代開銷（兩倍）；
2. Stream并行迭代的性能比串行迭代和外部迭代都好，

并行迭代性能跟可利用的核數有關，上圖中的并行迭代使用了全部12個核，為考察使用核數對性能的影響，我們專門測驗了不同核數下的Stream并行迭代效果：

分析，對于基本型別：

1. 使用Stream并行API在單核情況下性能很差，比Stream串行API的性能還差；
2. 隨著使用核數的增加，Stream并行效果逐漸變好，比使用for回圈外部迭代的性能還好，

以上兩個測驗說明，對于基本型別的簡單迭代，Stream串行迭代性能更差，但多核情況下Stream迭代時性能較好，

實驗二 物件迭代

再來看物件的迭代效果，

測驗內容：找出字串串列中最小的元素（自然順序），對比for回圈外部迭代和Stream API內部迭代性能，

測驗程式StringTest，測驗結果如下圖：

結果分析如下：

1. 對于物件型別Stream串行迭代的性能開銷仍然高于外部迭代開銷（1.5倍），但差距沒有基本型別那么大，
2. Stream并行迭代的性能比串行迭代和外部迭代都好，

再來單獨考察Stream并行迭代效果：

分析，對于物件型別：

1. 使用Stream并行API在單核情況下性能比for回圈外部迭代差；
2. 隨著使用核數的增加，Stream并行效果逐漸變好，多核帶來的效果明顯，

以上兩個測驗說明，對于物件型別的簡單迭代，Stream串行迭代性能更差，但多核情況下Stream迭代時性能較好，

實驗三 復雜物件歸約

從實驗一、二的結果來看，Stream串行執行的效果都比外部迭代差（很多），是不是說明Stream真的不行了？先別下結論，我們再來考察一下更復雜的操作，

測驗內容：給定訂單串列，統計每個用戶的總交易額，對比使用外部迭代手動實作和Stream API之間的性能，

我們將訂單簡化為<userName, price, timeStamp>構成的元組，并用Order物件來表示，測驗程式ReductionTest，測驗結果如下圖：

分析，對于復雜的歸約操作：

1. Stream API的性能普遍好于外部手動迭代，并行Stream效果更佳；

再來考察并行度對并行效果的影響，測驗結果如下：

分析，對于復雜的歸約操作：

1. 使用Stream并行歸約在單核情況下性能比串行歸約以及手動歸約都要差，簡單說就是最差的；
2. 隨著使用核數的增加，Stream并行效果逐漸變好，多核帶來的效果明顯，

以上兩個實驗說明，對于復雜的歸約操作，Stream串行歸約效果好于手動歸約，在多核情況下，并行歸約效果更佳，我們有理由相信，對于其他復雜的操作，Stream API也能表現出相似的性能表現，

結論

上述三個實驗的結果可以總結如下：

1. 對于簡單操作，比如最簡單的遍歷，Stream串行API性能明顯差于顯示迭代，但并行的Stream API能夠發揮多核特性，
2. 對于復雜操作，Stream串行API性能可以和手動實作的效果匹敵，在并行執行時Stream API效果遠超手動實作，

所以，如果出于性能考慮，1. 對于簡單操作推薦使用外部迭代手動實作，2. 對于復雜操作，推薦使用Stream API， 3. 在多核情況下，推薦使用并行Stream API來發揮多核優勢，4.單核情況下不建議使用并行Stream API，

參考文章

1. JavaLambdaInternals/6-Stream Pipelines.md
2. JavaLambdaInternals/8-Stream Performance.md
3. 極客時間-Java性能調優實戰/06.Stream如何提高遍歷集合效率？

公眾號

coding 筆記、點滴記錄，以后的文章也會同步到公眾號（Coding Insight）中，希望大家關注^_

代碼和思維導圖在 GitHub 專案中，歡迎大家 star！

標籤：Java

上一篇：dubbo原始碼決議-服務發布

下一篇：2020最新總結大廠Java高頻面試題（含答案決議）