轉自：http://lvheyang.com/?p=87

這兩天組內的小夥伴在學習Java8，推廣在新專案內使用新特性。正好看到了Stream 帶來的遍歷的多執行緒併發：
    

Arrays.asList(1, 2, 3, 4, 5, 6, 7, 9, 8, 0, 1)
        .stream()
        .parallel()
        .collect(Collectors.groupingBy(x -> x % 10))
        .forEach((x, y) -> System.out.println(x + ":" + y));

和小夥伴一起試著用各種玩法玩了一下Java8的函數語言程式設計特性之後，感嘆到這樣子併發計算真的是越來越簡單了的。

但是深入思考之後就會很自然的想到一個問題，這個過程中，我們並沒有顯示的告訴Stream，我們需要多少個執行緒進行平行計算？我們能否複用之前的執行緒進行計算？

帶著這個問題我們先打開了VisualVM，檢視一下我們執行這樣一個任務會啟動多少個執行緒？



我們可以看到預設的parallel計算啟動了三個執行緒進行並行。這三個執行緒是怎麼來的呢？抱著這個問題，我們來參考一下Jdk8的原始碼，來看看它是如何設定這個值的。

我們知道Stream 是一個惰性求值的系統（如何進行惰性求值，我會在另一篇部落格中進行分析），那麼我們只需要找它最後求值的過程，看它是怎樣進行求值的就可以了。在AbstractPipeline 這個類裡面我們找到了Stream 計算的最終求值過程的預設實現：

/**
 * Evaluate the pipeline with a terminal operation to produce a result.
 *
 * @param <R> the type of result
 * @param terminalOp the terminal operation to be applied to the pipeline.
 * @return the result
 */
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()));
}
 

在15行，我們可以看到，在求值的時候會檢查平行計算的標誌位，如果標誌了平行計算的話，我們就會並行求值，反之則會序列求值。我們可以進一步進入並行求值的邏輯中，這是一個TerminalOp的預設介面方法，預設實現就是直接呼叫序列求值，在FindOp、ForEachOp、MatchOp 和 ReduceOp 中得到了覆蓋。

@Override
public <P_IN> O evaluateParallel(PipelineHelper<T> helper,
                                 Spliterator<P_IN> spliterator) {
    return new FindTask<>(this, helper, spliterator).invoke();
}
 

如FindOp的程式碼示例，這四個操作都是建立一個Task的示例，然後執行invoke方法。這些Task的繼承關係如圖：



可以看出所有的Task 都繼承自Jdk7 中引入的ForkJoin 並行框架的ForkJoinTask。所以我們可以看出Stream 的並行是依賴於ForkJoin 框架的。以AbstractTask 為例我們看看它是如何進行平行計算的：

@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();
}

ReduceTask(ReduceTask<P_IN, P_OUT, R, S> parent,
           Spliterator spliterator) {
    super(parent, spliterator);
    this.op = parent.op;
}

@Override
protected ReduceTask<P_IN, P_OUT, R, S> makeChild(Spliterator spliterator) {
    return new ReduceTask<>(this, spliterator);
}

這裡面的主要邏輯就是

    先呼叫當前splititerator 方法的estimateSize 方法，預估這個分片中的資料量
    根據預估的資料量獲取最小處理單元的大小閾值，即當資料量已經小於這個閾值的時候進行計算，否則進行fork 將任務劃分成更小的資料塊，進行求解。這裡值得注意的是，getTargetSize 在第一次呼叫的時候會設定：
        預測資料量大小 / (預設併發度 * 4) 的結果作為最小執行單元的數量（配置的預設值是cpu 數 – 1，可以通過java.util.concurrent.ForkJoinPool.common.parallelism設定）
    如果當前分片大小仍然大於處理資料單元的閾值，且分片繼續嘗試切分成功，那麼就繼續切分，分別將左右分片的任務建立為新的Task，並且將當前的任務關聯為兩個新任務的父級任務（邏輯在makeChild 裡面）
    先後對左右子節點的任務進行fork，對另外的分割槽進行分解。同時設定pending 為1，這代表一個task 實際上只會有一個等待的子節點（被fork）。
    當任務已經分解到足夠小的時候退出迴圈，嘗試進行結束。呼叫子類實現的doLeaf方法，完成最小計算單元的計算任務，並設定到當前任務的localResult中
    呼叫tryComplete 方法進行最終任務的掃尾工作，如果該任務pending 值不等於0，則原子的減1，如果已經等於0，說明任務都已經完成，則呼叫onCompletion 回撥，如果該任務是葉子任務，則直接銷燬中間資料結束；如果是中間節點會將左右子節點的結果進行合併
    檢查如果這個任務已經沒有父級任務了，則將該任務置為正常結束，如果還有則嘗試遞迴的去呼叫父級節點的onCompletion回撥，逐級進行任務的合併。

public final void tryComplete() {
    CountedCompleter<?> a = this, s = a;
    for (int c;;) {
        if ((c = a.pending) == 0) {
            a.onCompletion(s);
            if ((a = (s = a).completer) == null) {
                s.quietlyComplete();
                return;
            }
        }
        else if (U.compareAndSwapInt(a, PENDING, c, c - 1))
            return;
    }
}

說了這麼多，大家也基本理解了Stream 的實現原理了。其實本質上就是在ForkJoin上進行了一層封裝，將Stream 不斷嘗試分解成更小的split，然後使用fork/join 框架分而治之。

所以我們以往關於Fork/Join 的經驗也都可以派上用場，可以解答之前我們的幾個疑問：

    我在visualvm 中看到的 parallize 的3個執行緒是怎麼來的？
        答：由於 taskToFork.fork() 呼叫，parallize使用了預設的ForkJoinPool.common 預設的一個靜態執行緒池，這個執行緒池的預設執行緒個數是cpu 數量-1。由於我的程式碼是執行在四個邏輯核心的MacBook 上，所以這裡的執行緒個數為3。如下面程式碼和註釋所示：

if (parallelism < 0 && // default 1 less than #cores
    (parallelism = Runtime.getRuntime().availableProcessors() - 1) <= 0)
    parallelism = 1;
if (parallelism > MAX_CAP)
    parallelism = MAX_CAP;

    如何控制parallize 的執行緒數？
        答：我們可以自己構建一個ForkJoinPool，向其中提交一個parallize 任務，可以做到控制併發度。如以下示例程式碼：我們將之前的stream 的過程構造成一個runnable 的lambda 匿名函式 ()-> {…}。提交至執行緒池中，就可以按照我們想要的併發度進行計算了。

ForkJoinPool pool = new ForkJoinPool(2);
ret = pool.submit(() -> {
    return LongStream.range(1, 50 * 1024 * 1024).boxed().collect(Collectors.toList())
            .stream()
            .parallel()
            .map(x -> x * 2)
            .filter(x -> x < 1500)
            .reduce((x,y) -> x+y)
            .get();
}).get();

接下來打算繼續深入這兩個很有意思的問題：

    深入介紹一下Stream的惰性求值過程，最好能跟Scala 的Stream 實現進行比較：Java8 Stream 惰性求值實現分析 – 驢和羊

    深入介紹ForkJoin 的底層實現，包括它是如何進行執行緒排程和cache line sharing 優化的

參考文獻：