Java Stream 原始碼分析

阿新 • • 發佈：2020-12-07

Yano_nankai 公眾號：Coding Insight，專注 Java 技術

Java Stream 原始碼分析

前言

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

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次求出平均時間作為執行時間。

實驗一基本型別迭代

測試內容：找出整型陣列中的最小值。對比for迴圈外部迭代和Stream API內部迭代效能。

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

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

對於基本型別Stream序列迭代的效能開銷明顯高於外部迭代開銷（兩倍）；
Stream並行迭代的效能比序列迭代和外部迭代都好。

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

分析，對於基本型別：

使用Stream並行API在單核情況下效能很差，比Stream序列API的效能還差；
隨著使用核數的增加，Stream並行效果逐漸變好，比使用for迴圈外部迭代的效能還好。

以上兩個測試說明，對於基本型別的簡單迭代，Stream序列迭代效能更差，但多核情況下Stream迭代時效能較好。

實驗二物件迭代

再來看物件的迭代效果。

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

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

結果分析如下：

對於物件型別Stream序列迭代的效能開銷仍然高於外部迭代開銷（1.5倍），但差距沒有基本型別那麼大。
Stream並行迭代的效能比序列迭代和外部迭代都好。

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

分析，對於物件型別：

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

以上兩個測試說明，對於物件型別的簡單迭代，Stream序列迭代效能更差，但多核情況下Stream迭代時效能較好。

實驗三複雜物件歸約

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

測試內容：給定訂單列表，統計每個使用者的總交易額。對比使用外部迭代手動實現和Stream API之間的效能。

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

分析，對於複雜的歸約操作：

Stream API的效能普遍好於外部手動迭代，並行Stream效果更佳；

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

分析，對於複雜的歸約操作：

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

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

結論

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

對於簡單操作，比如最簡單的遍歷，Stream序列API效能明顯差於顯示迭代，但並行的Stream API能夠發揮多核特性。
對於複雜操作，Stream序列API效能可以和手動實現的效果匹敵，在並行執行時Stream API效果遠超手動實現。

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

Java Stream 原始碼分析

前言

操作分類

原始碼結構

BaseStream

Stream

ReferencePipeline

Sink

操作疊加

載入資料來源

中間操作

終結操作

並行處理

並行錯誤的使用方法

並行正確的使用方法

效能

測試方法和測試資料

實驗一 基本型別迭代

實驗二 物件迭代

實驗三 複雜物件歸約

結論

相關推薦

實驗一基本型別迭代

實驗二物件迭代

實驗三複雜物件歸約