前面我們已經學會如何使用Stream API，用起來真的很爽，但簡潔的方法下面似乎隱藏著無盡的祕密，如此強大的API是如何實現的呢？Pipeline是怎麼執行的，每次方法呼叫都會導致一次迭代嗎？自動並行又是怎麼做到的，執行緒個數是多少？本節我們學習Stream流水線的原理，這是Stream實現的關鍵所在。

首先回顧一下容器執行Lambda表示式的方式，以ArrayList.forEach()方法為例，具體程式碼如下：

// ArrayList.forEach()
public void forEach(Consumer<? super E> action) {
    ...
    for (int i=0; modCount == expectedModCount && i < size; i++) {
        action.accept(elementData[i]);// 回撥方法
    }
    ...
}
 

我們看到ArrayList.forEach()方法的主要邏輯就是一個for迴圈，在該for迴圈裡不斷呼叫action.accept()回撥方法完成對元素的遍歷。這完全沒有什麼新奇之處，回撥方法在Java GUI的監聽器中廣泛使用。Lambda表示式的作用就是相當於一個回撥方法，這很好理解。

Stream API中大量使用Lambda表示式作為回撥方法，但這並不是關鍵。理解Stream我們更關心的是另外兩個問題：流水線和自動並行。使用Stream或許很容易寫入如下形式的程式碼：

int longestStringLengthStartingWithA
        = strings.stream()
              .filter(s -> s.startsWith("A"))
              .mapToInt(String::length)
              .max();
 

上述程式碼求出以字母A開頭的字串的最大長度，一種直白的方式是為每一次函式呼叫都執一次迭代，這樣做能夠實現功能，但效率上肯定是無法接受的。類庫的實現著使用流水線（Pipeline）的方式巧妙的避免了多次迭代，其基本思想是在一次迭代中儘可能多的執行使用者指定的操作。為講解方便我們彙總了Stream的所有操作。

Stream上的所有操作分為兩類：中間操作和結束操作，中間操作只是一種標記，只有結束操作才會觸發實際計算。中間操作又可以分為無狀態的(Stateless)和有狀態的(Stateful)，無狀態中間操作是指元素的處理不受前面元素的影響，而有狀態的中間操作必須等到所有元素處理之後才知道最終結果，比如排序是有狀態操作，在讀取所有元素之前並不能確定排序結果；結束操作又可以分為短路操作和非短路操作，短路操作是指不用處理全部元素就可以返回結果，比如找到第一個滿足條件的元素。之所以要進行如此精細的劃分，是因為底層對每一種情況的處理方式不同。

一種直白的實現方式

仍然考慮上述求最長字串的程式，一種直白的流水線實現方式是為每一次函式呼叫都執一次迭代，並將處理中間結果放到某種資料結構中（比如陣列，容器等）。具體說來，就是呼叫filter()方法後立即執行，選出所有以A開頭的字串並放到一個列表list1中，之後讓list1傳遞給mapToInt()方法並立即執行，生成的結果放到list2中，最後遍歷list2找出最大的數字作為最終結果。程式的執行流程如如所示：

這樣做實現起來非常簡單直觀，但有兩個明顯的弊端：

1. 迭代次數多。迭代次數跟函式呼叫的次數相等。
2. 頻繁產生中間結果。每次函式呼叫都產生一次中間結果，儲存開銷無法接受。

這些弊端使得效率底下，根本無法接受。如果不使用Stream API我們都知道上述程式碼該如何在一次迭代中完成，大致是如下形式：

int longest = 0;
for(String str : strings){
    if(str.startsWith("A")){// 1. filter(), 保留以A開頭的字串
        int len = str.length();// 2. mapToInt(), 轉換成長度
        longest = Math.max(len, longest);// 3. max(), 保留最長的長度
    }
}

採用這種方式我們不但減少了迭代次數，也避免了儲存中間結果，顯然這就是流水線，因為我們把三個操作放在了一次迭代當中。只要我們事先知道使用者意圖，總是能夠採用上述方式實現跟Stream API等價的功能，但問題是Stream類庫的設計者並不知道使用者的意圖是什麼。如何在無法假設使用者行為的前提下實現流水線，是類庫的設計者要考慮的問題。

Stream流水線解決方案

我們大致能夠想到，應該採用某種方式記錄使用者每一步的操作，當用戶呼叫結束操作時將之前記錄的操作疊加到一起在一次迭代中全部執行掉。沿著這個思路，有幾個問題需要解決：

1. 使用者的操作如何記錄？
2. 操作如何疊加？
3. 疊加之後的操作如何執行？
4. 執行後的結果（如果有）在哪裡？

>> 操作如何記錄

注意這裡使用的是“操作(operation)”一詞，指的是“Stream中間操作”的操作，很多Stream操作會需要一個回撥函式（Lambda表示式），因此一個完整的操作是<資料來源，操作，回撥函式>構成的三元組。Stream中使用Stage的概念來描述一個完整的操作，並用某種例項化後的PipelineHelper來代表Stage，將具有先後順序的各個Stage連到一起，就構成了整個流水線。跟Stream相關類和介面的繼承關係圖示。

還有IntPipeline, LongPipeline, DoublePipeline沒在圖中畫出，這三個類專門為三種基本型別（不是包裝型別）而定製的，跟ReferencePipeline是並列關係。圖中Head用於表示第一個Stage，即呼叫呼叫諸如Collection.stream()方法產生的Stage，很顯然這個Stage裡不包含任何操作；StatelessOp和StatefulOp分別表示無狀態和有狀態的Stage，對應於無狀態和有狀態的中間操作。

Stream流水線組織結構示意圖如下：

圖中通過Collection.stream()方法得到Head也就是stage0，緊接著呼叫一系列的中間操作，不斷產生新的Stream。這些Stream物件以雙向連結串列的形式組織在一起，構成整個流水線，由於每個Stage都記錄了前一個Stage和本次的操作以及回撥函式，依靠這種結構就能建立起對資料來源的所有操作。這就是Stream記錄操作的方式。

>> 操作如何疊加

以上只是解決了操作記錄的問題，要想讓流水線起到應有的作用我們需要一種將所有操作疊加到一起的方案。你可能會覺得這很簡單，只需要從流水線的head開始依次執行每一步的操作（包括回撥函式）就行了。這聽起來似乎是可行的，但是你忽略了前面的Stage並不知道後面Stage到底執行了哪種操作，以及回撥函式是哪種形式。換句話說，只有當前Stage本身才知道該如何執行自己包含的動作。這就需要有某種協議來協調相鄰Stage之間的呼叫關係。

這種協議由Sink介面完成，Sink介面包含的方法如下表所示：

有了上面的協議，相鄰Stage之間呼叫就很方便了，每個Stage都會將自己的操作封裝到一個Sink裡，前一個Stage只需呼叫後一個Stage的accept()方法即可，並不需要知道其內部是如何處理的。當然對於有狀態的操作，Sink的begin()和end()方法也是必須實現的。比如Stream.sorted()是一個有狀態的中間操作，其對應的Sink.begin()方法可能建立一個乘放結果的容器，而accept()方法負責將元素新增到該容器，最後end()負責對容器進行排序。對於短路操作，Sink.cancellationRequested()也是必須實現的，比如Stream.findFirst()是短路操作，只要找到一個元素，cancellationRequested()就應該返回true，以便呼叫者儘快結束查詢。Sink的四個介面方法常常相互協作，共同完成計算任務。實際上Stream API內部實現的的本質，就是如何過載Sink的這四個介面方法。

有了Sink對操作的包裝，Stage之間的呼叫問題就解決了，執行時只需要從流水線的head開始對資料來源依次呼叫每個Stage對應的Sink.{begin(), accept(), cancellationRequested(), end()}方法就可以了。一種可能的Sink.accept()方法流程是這樣的：

void accept(U u){
    1. 使用當前Sink包裝的回撥函式處理u
    2. 將處理結果傳遞給流水線下游的Sink
}

Sink介面的其他幾個方法也是按照這種[處理->轉發]的模型實現。下面我們結合具體例子看看Stream的中間操作是如何將自身的操作包裝成Sink以及Sink是如何將處理結果轉發給下一個Sink的。先看Stream.map()方法：

// Stream.map()，呼叫該方法將產生一個新的Stream
public final <R> Stream<R> map(Function<? super P_OUT, ? extends R> mapper) {
    ...
    return new StatelessOp<P_OUT, R>(this, StreamShape.REFERENCE,
                                 StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
        @Override /*opWripSink()方法返回由回撥函式包裝而成Sink*/
        Sink<P_OUT> opWrapSink(int flags, Sink<R> downstream) {
            return new Sink.ChainedReference<P_OUT, R>(downstream) {
                @Override
                public void accept(P_OUT u) {
                    R r = mapper.apply(u);// 1. 使用當前Sink包裝的回撥函式mapper處理u
                    downstream.accept(r);// 2. 將處理結果傳遞給流水線下游的Sink
                }
            };
        }
    };
}

上述程式碼看似複雜，其實邏輯很簡單，就是將回調函式mapper包裝到一個Sink當中。由於Stream.map()是一個無狀態的中間操作，所以map()方法返回了一個StatelessOp內部類物件（一個新的Stream），呼叫這個新Stream的opWripSink()方法將得到一個包裝了當前回調函式的Sink。

再來看一個複雜一點的例子。Stream.sorted()方法將對Stream中的元素進行排序，顯然這是一個有狀態的中間操作，因為讀取所有元素之前是沒法得到最終順序的。拋開模板程式碼直接進入問題本質，sorted()方法是如何將操作封裝成Sink的呢？sorted()一種可能封裝的Sink程式碼如下：

// Stream.sort()方法用到的Sink實現
class RefSortingSink<T> extends AbstractRefSortingSink<T> {
    private ArrayList<T> list;// 存放用於排序的元素
    RefSortingSink(Sink<? super T> downstream, Comparator<? super T> comparator) {
        super(downstream, comparator);
    }
    @Override
    public void begin(long size) {
        ...
        // 建立一個存放排序元素的列表
        list = (size >= 0) ? new ArrayList<T>((int) size) : new ArrayList<T>();
    }
    @Override
    public void end() {
        list.sort(comparator);// 只有元素全部接收之後才能開始排序
        downstream.begin(list.size());
        if (!cancellationWasRequested) {// 下游Sink不包含短路操作
            list.forEach(downstream::accept);// 2. 將處理結果傳遞給流水線下游的Sink
        }
        else {// 下游Sink包含短路操作
            for (T t : list) {// 每次都呼叫cancellationRequested()詢問是否可以結束處理。
                if (downstream.cancellationRequested()) break;
                downstream.accept(t);// 2. 將處理結果傳遞給流水線下游的Sink
            }
        }
        downstream.end();
        list = null;
    }
    @Override
    public void accept(T t) {
        list.add(t);// 1. 使用當前Sink包裝動作處理t，只是簡單的將元素新增到中間列表當中
    }
}

上述程式碼完美的展現了Sink的四個介面方法是如何協同工作的：

1. 首先beging()方法告訴Sink參與排序的元素個數，方便確定中間結果容器的的大小；
2. 之後通過accept()方法將元素新增到中間結果當中，最終執行時呼叫者會不斷呼叫該方法，直到遍歷所有元素；
3. 最後end()方法告訴Sink所有元素遍歷完畢，啟動排序步驟，排序完成後將結果傳遞給下游的Sink；
4. 如果下游的Sink是短路操作，將結果傳遞給下游時不斷詢問下游cancellationRequested()是否可以結束處理。

>> 疊加之後的操作如何執行

Sink完美封裝了Stream每一步操作，並給出了[處理->轉發]的模式來疊加操作。這一連串的齒輪已經咬合，就差最後一步撥動齒輪啟動執行。是什麼啟動這一連串的操作呢？也許你已經想到了啟動的原始動力就是結束操作(Terminal Operation)，一旦呼叫某個結束操作，就會觸發整個流水線的執行。

結束操作之後不能再有別的操作，所以結束操作不會建立新的流水線階段(Stage)，直觀的說就是流水線的連結串列不會在往後延伸了。結束操作會建立一個包裝了自己操作的Sink，這也是流水線中最後一個Sink，這個Sink只需要處理資料而不需要將結果傳遞給下游的Sink（因為沒有下游）。對於Sink的[處理->轉發]模型，結束操作的Sink就是呼叫鏈的出口。

我們再來考察一下上游的Sink是如何找到下游Sink的。一種可選的方案是在PipelineHelper中設定一個Sink欄位，在流水線中找到下游Stage並訪問Sink欄位即可。但Stream類庫的設計者沒有這麼做，而是設定了一個Sink AbstractPipeline.opWrapSink(int flags, Sink downstream)方法來得到Sink，該方法的作用是返回一個新的包含了當前Stage代表的操作以及能夠將結果傳遞給downstream的Sink物件。為什麼要產生一個新物件而不是返回一個Sink欄位？這是因為使用opWrapSink()可以將當前操作與下游Sink（上文中的downstream引數）結合成新Sink。試想只要從流水線的最後一個Stage開始，不斷呼叫上一個Stage的opWrapSink()方法直到最開始（不包括stage0，因為stage0代表資料來源，不包含操作），就可以得到一個代表了流水線上所有操作的Sink，用程式碼表示就是這樣：

// AbstractPipeline.wrapSink()
// 從下游向上遊不斷包裝Sink。如果最初傳入的sink代表結束操作，
// 函式返回時就可以得到一個代表了流水線上所有操作的Sink。
final <P_IN> Sink<P_IN> wrapSink(Sink<E_OUT> sink) {
    ...
    for (AbstractPipeline p=AbstractPipeline.this; p.depth > 0; p=p.previousStage) {
        sink = p.opWrapSink(p.previousStage.combinedFlags, sink);
    }
    return (Sink<P_IN>) sink;
}

現在流水線上從開始到結束的所有的操作都被包裝到了一個Sink裡，執行這個Sink就相當於執行整個流水線，執行Sink的程式碼如下：

// AbstractPipeline.copyInto(), 對spliterator代表的資料執行wrappedSink代表的操作。
final <P_IN> void copyInto(Sink<P_IN> wrappedSink, Spliterator<P_IN> spliterator) {
    ...
    if (!StreamOpFlag.SHORT_CIRCUIT.isKnown(getStreamAndOpFlags())) {
        wrappedSink.begin(spliterator.getExactSizeIfKnown());// 通知開始遍歷
        spliterator.forEachRemaining(wrappedSink);// 迭代
        wrappedSink.end();// 通知遍歷結束
    }
    ...
}

上述程式碼首先呼叫wrappedSink.begin()方法告訴Sink資料即將到來，然後呼叫spliterator.forEachRemaining()方法對資料進行迭代（Spliterator是容器的一種迭代器，參閱），最後呼叫wrappedSink.end()方法通知Sink資料處理結束。邏輯如此清晰。

>> 執行後的結果在哪裡

最後一個問題是流水線上所有操作都執行後，使用者所需要的結果（如果有）在哪裡？首先要說明的是不是所有的Stream結束操作都需要返回結果，有些操作只是為了使用其副作用(Side-effects)，比如使用Stream.forEach()方法將結果打印出來就是常見的使用副作用的場景（事實上，除了列印之外其他場景都應避免使用副作用），對於真正需要返回結果的結束操作結果存在哪裡呢？

特別說明：副作用不應該被濫用，也許你會覺得在Stream.forEach()裡進行元素收集是個不錯的選擇，就像下面程式碼中那樣，但遺憾的是這樣使用的正確性和效率都無法保證，因為Stream可能會並行執行。大多數使用副作用的地方都可以使用歸約操作更安全和有效的完成。

// 錯誤的收集方式
ArrayList<String> results = new ArrayList<>();
stream.filter(s -> pattern.matcher(s).matches())
      .forEach(s -> results.add(s));  // Unnecessary use of side-effects!
// 正確的收集方式
List<String>results =
     stream.filter(s -> pattern.matcher(s).matches())
             .collect(Collectors.toList());  // No side-effects!

回到流水線執行結果的問題上來，需要返回結果的流水線結果存在哪裡呢？這要分不同的情況討論，下表給出了各種有返回結果的Stream結束操作。

1. 對於表中返回boolean或者Optional的操作（Optional是存放 一個 值的容器）的操作，由於值返回一個值，只需要在對應的Sink中記錄這個值，等到執行結束時返回就可以了。
2. 對於歸約操作，最終結果放在使用者呼叫時指定的容器中（容器型別通過收集器指定）。collect(), reduce(), max(), min()都是歸約操作，雖然max()和min()也是返回一個Optional，但事實上底層是通過呼叫reduce()方法實現的。
3. 對於返回是陣列的情況，毫無疑問的結果會放在陣列當中。這麼說當然是對的，但在最終返回陣列之前，結果其實是儲存在一種叫做Node的資料結構中的。Node是一種多叉樹結構，元素儲存在樹的葉子當中，並且一個葉子節點可以存放多個元素。這樣做是為了並行執行方便。關於Node的具體結構，我們會在下一節探究Stream如何並行執行時給出詳細說明。

結語

本文詳細介紹了Stream流水線的組織方式和執行過程，學習本文將有助於理解原理並寫出正確的Stream程式碼，同時打消你對Stream API效率方面的顧慮。如你所見，Stream API實現如此巧妙，即使我們使用外部迭代手動編寫等價程式碼，也未必更加高效。