摘要：數(shù)據(jù)流教程原文譯者飛龍協(xié)議這個(gè)示例驅(qū)動(dòng)的教程是數(shù)據(jù)流的深入總結(jié)。但是的數(shù)據(jù)流是完全不同的東西。數(shù)據(jù)流是單體，并且在函數(shù)式編程中起到重要作用。列表上的所有流式操作請(qǐng)見數(shù)據(jù)流的。基本的數(shù)據(jù)流使用特殊的表達(dá)式，例如，而不是，而不是。

原文：Java 8 Stream Tutorial 
譯者：飛龍  
協(xié)議：CC BY-NC-SA 4.0

這個(gè)示例驅(qū)動(dòng)的教程是Java8數(shù)據(jù)流（Stream）的深入總結(jié)。當(dāng)我第一次看到StreamAPI時(shí)，我非常疑惑，因?yàn)樗犉饋?lái)和Java IO的InputStream 和 OutputStream一樣。但是Java8的數(shù)據(jù)流是完全不同的東西。數(shù)據(jù)流是單體（Monad），并且在Java8函數(shù)式編程中起到重要作用。

在函數(shù)式編程中，單體是一個(gè)結(jié)構(gòu)，表示定義為步驟序列的計(jì)算。單體結(jié)構(gòu)的類型定義了它對(duì)鏈?zhǔn)讲僮鳎蚓哂邢嗤愋偷那短缀瘮?shù)的含義。

這個(gè)教程教給你如何使用Java8數(shù)據(jù)流，以及如何使用不同種類的可用的數(shù)據(jù)流操作。你將會(huì)學(xué)到處理次序以及流操作的次序如何影響運(yùn)行時(shí)效率。這個(gè)教程也會(huì)詳細(xì)講解更加強(qiáng)大的流操作，reduce、collect和flatMap。最后，這個(gè)教程會(huì)深入探討并行流。

如果你還不熟悉Java8的lambda表達(dá)式，函數(shù)式接口和方法引用，你可能需要在開始這一章之前，首先閱讀我的Java8教程。

更新 - 我現(xiàn)在正在編寫用于瀏覽器的Java8數(shù)據(jù)流API的JavaScript實(shí)現(xiàn)。如果你對(duì)此感興趣，請(qǐng)?jiān)贕ithub上訪問(wèn)Stream.js。非常期待你的反饋。

數(shù)據(jù)流表示元素的序列，并支持不同種類的操作來(lái)執(zhí)行元素上的計(jì)算：

List myList =
    Arrays.asList("a1", "a2", "b1", "c2", "c1");

myList
    .stream()
    .filter(s -> s.startsWith("c"))
    .map(String::toUpperCase)
    .sorted()
    .forEach(System.out::println);

// C1
// C2

數(shù)據(jù)流操作要么是銜接操作，要么是終止操作。銜接操作返回?cái)?shù)據(jù)流，所以我們可以把多個(gè)銜接操作不使用分號(hào)來(lái)鏈接到一起。終止操作無(wú)返回值，或者返回一個(gè)不是流的結(jié)果。在上面的例子中，filter、map和sorted都是銜接操作，而forEach是終止操作。列表上的所有流式操作請(qǐng)見數(shù)據(jù)流的Javadoc。你在上面例子中看到的這種數(shù)據(jù)流的鏈?zhǔn)讲僮饕步凶鞑僮髁魉€。

多數(shù)數(shù)據(jù)流操作都接受一些lambda表達(dá)式參數(shù)，函數(shù)式接口用來(lái)指定操作的具體行為。這些操作的大多數(shù)必須是無(wú)干擾而且是無(wú)狀態(tài)的。它們是什么意思呢？

當(dāng)一個(gè)函數(shù)不修改數(shù)據(jù)流的底層數(shù)據(jù)源，它就是無(wú)干擾的。例如，在上面的例子中，沒(méi)有任何lambda表達(dá)式通過(guò)添加或刪除集合元素修改myList。

當(dāng)一個(gè)函數(shù)的操作的執(zhí)行是確定性的，它就是無(wú)狀態(tài)的。例如，在上面的例子中，沒(méi)有任何lambda表達(dá)式依賴于外部作用域中任何在操作過(guò)程中可變的變量或狀態(tài)。

數(shù)據(jù)流可以從多種數(shù)據(jù)源創(chuàng)建，尤其是集合。List和Set支持新方法stream() 和 parallelStream()，來(lái)創(chuàng)建串行流或并行流。并行流能夠在多個(gè)線程上執(zhí)行操作，它們會(huì)在之后的章節(jié)中講到。我們現(xiàn)在來(lái)看看串行流：

Arrays.asList("a1", "a2", "a3")
    .stream()
    .findFirst()
    .ifPresent(System.out::println);  // a1

在對(duì)象列表上調(diào)用stream()方法會(huì)返回一個(gè)通常的對(duì)象流。但是我們不需要?jiǎng)?chuàng)建一個(gè)集合來(lái)創(chuàng)建數(shù)據(jù)流，就像下面那樣：

Stream.of("a1", "a2", "a3")
    .findFirst()
    .ifPresent(System.out::println);  // a1

只要使用Stream.of()，就可以從一系列對(duì)象引用中創(chuàng)建數(shù)據(jù)流。

除了普通的對(duì)象數(shù)據(jù)流，Java8還自帶了特殊種類的流，用于處理基本數(shù)據(jù)類型int、long 和 double。你可能已經(jīng)猜到了它是IntStream、LongStream 和 DoubleStream。

IntStream可以使用IntStream.range()替換通常的for循環(huán)：

IntStream.range(1, 4)
    .forEach(System.out::println);

// 1
// 2
// 3

所有這些基本數(shù)據(jù)流都像通常的對(duì)象數(shù)據(jù)流一樣，但有一些不同。基本的數(shù)據(jù)流使用特殊的lambda表達(dá)式，例如，IntFunction而不是Function，IntPredicate而不是Predicate。而且基本數(shù)據(jù)流支持額外的聚合終止操作sum()和average()：

Arrays.stream(new int[] {1, 2, 3})
    .map(n -> 2 * n + 1)
    .average()
    .ifPresent(System.out::println);  // 5.0

有時(shí)需要將通常的對(duì)象數(shù)據(jù)流轉(zhuǎn)換為基本數(shù)據(jù)流，或者相反。出于這種目的，對(duì)象數(shù)據(jù)流支持特殊的映射操作mapToInt()、mapToLong() 和 mapToDouble()：

Stream.of("a1", "a2", "a3")
    .map(s -> s.substring(1))
    .mapToInt(Integer::parseInt)
    .max()
    .ifPresent(System.out::println);  // 3

基本數(shù)據(jù)流可以通過(guò)maoToObj()轉(zhuǎn)換為對(duì)象數(shù)據(jù)流：

IntStream.range(1, 4)
    .mapToObj(i -> "a" + i)
    .forEach(System.out::println);

// a1
// a2
// a3

下面是組合示例：浮點(diǎn)數(shù)據(jù)流首先映射為整數(shù)數(shù)據(jù)流，之后映射為字符串的對(duì)象數(shù)據(jù)流：

Stream.of(1.0, 2.0, 3.0)
    .mapToInt(Double::intValue)
    .mapToObj(i -> "a" + i)
    .forEach(System.out::println);

// a1
// a2
// a3

既然我們已經(jīng)了解了如何創(chuàng)建并使用不同種類的數(shù)據(jù)流，讓我們深入了解數(shù)據(jù)流操作在背后如何執(zhí)行吧。

銜接操作的一個(gè)重要特性就是延遲性。觀察下面沒(méi)有終止操作的例子：

Stream.of("d2", "a2", "b1", "b3", "c")
    .filter(s -> {
        System.out.println("filter: " + s);
        return true;
    });

執(zhí)行這段代碼時(shí)，不向控制臺(tái)打印任何東西。這是因?yàn)殂暯硬僮髦辉诮K止操作調(diào)用時(shí)被執(zhí)行。

讓我們通過(guò)添加終止操作forEach來(lái)擴(kuò)展這個(gè)例子：

Stream.of("d2", "a2", "b1", "b3", "c")
    .filter(s -> {
        System.out.println("filter: " + s);
        return true;
    })
    .forEach(s -> System.out.println("forEach: " + s));

執(zhí)行這段代碼會(huì)得到如下輸出：

filter:  d2
forEach: d2
filter:  a2
forEach: a2
filter:  b1
forEach: b1
filter:  b3
forEach: b3
filter:  c
forEach: c

結(jié)果的順序可能出人意料。原始的方法會(huì)在數(shù)據(jù)流的所有元素上，一個(gè)接一個(gè)地水平執(zhí)行所有操作。但是每個(gè)元素在調(diào)用鏈上垂直移動(dòng)。第一個(gè)字符串"d2"首先經(jīng)過(guò)filter然后是forEach，執(zhí)行完后才開始處理第二個(gè)字符串"a2"。

這種行為可以減少每個(gè)元素上所執(zhí)行的實(shí)際操作數(shù)量，就像我們?cè)谙聜€(gè)例子中看到的那樣：

Stream.of("d2", "a2", "b1", "b3", "c")
    .map(s -> {
        System.out.println("map: " + s);
        return s.toUpperCase();
    })
    .anyMatch(s -> {
        System.out.println("anyMatch: " + s);
        return s.startsWith("A");
    });

// map:      d2
// anyMatch: D2
// map:      a2
// anyMatch: A2

只要提供的數(shù)據(jù)元素滿足了謂詞，anyMatch操作就會(huì)返回true。對(duì)于第二個(gè)傳遞"A2"的元素，它的結(jié)果為真。由于數(shù)據(jù)流的鏈?zhǔn)秸{(diào)用是垂直執(zhí)行的，map這里只需要執(zhí)行兩次。所以map會(huì)執(zhí)行盡可能少的次數(shù)，而不是把所有元素都映射一遍。

下面的例子由兩個(gè)銜接操作map和filter，以及一個(gè)終止操作forEach組成。讓我們?cè)賮?lái)看看這些操作如何執(zhí)行：

Stream.of("d2", "a2", "b1", "b3", "c")
    .map(s -> {
        System.out.println("map: " + s);
        return s.toUpperCase();
    })
    .filter(s -> {
        System.out.println("filter: " + s);
        return s.startsWith("A");
    })
    .forEach(s -> System.out.println("forEach: " + s));

// map:     d2
// filter:  D2
// map:     a2
// filter:  A2
// forEach: A2
// map:     b1
// filter:  B1
// map:     b3
// filter:  B3
// map:     c
// filter:  C

就像你可能猜到的那樣，map和filter會(huì)對(duì)底層集合的每個(gè)字符串調(diào)用五次，而forEach只會(huì)調(diào)用一次。

如果我們調(diào)整操作順序，將filter移動(dòng)到調(diào)用鏈的頂端，就可以極大減少操作的執(zhí)行次數(shù):

Stream.of("d2", "a2", "b1", "b3", "c")
    .filter(s -> {
        System.out.println("filter: " + s);
        return s.startsWith("a");
    })
    .map(s -> {
        System.out.println("map: " + s);
        return s.toUpperCase();
    })
    .forEach(s -> System.out.println("forEach: " + s));

// filter:  d2
// filter:  a2
// map:     a2
// forEach: A2
// filter:  b1
// filter:  b3
// filter:  c

現(xiàn)在，map只會(huì)調(diào)用一次，所以操作流水線對(duì)于更多的輸入元素會(huì)執(zhí)行更快。在整合復(fù)雜的方法鏈時(shí)，要記住這一點(diǎn)。

讓我們通過(guò)添加額外的方法sorted來(lái)擴(kuò)展上面的例子：

Stream.of("d2", "a2", "b1", "b3", "c")
    .sorted((s1, s2) -> {
        System.out.printf("sort: %s; %s
", s1, s2);
        return s1.compareTo(s2);
    })
    .filter(s -> {
        System.out.println("filter: " + s);
        return s.startsWith("a");
    })
    .map(s -> {
        System.out.println("map: " + s);
        return s.toUpperCase();
    })
    .forEach(s -> System.out.println("forEach: " + s));

排序是一類特殊的銜接操作。它是有狀態(tài)的操作，因?yàn)槟阈枰谔幚碇斜４鏍顟B(tài)來(lái)對(duì)集合中的元素排序。

執(zhí)行這個(gè)例子會(huì)得到如下輸入：

sort:    a2; d2
sort:    b1; a2
sort:    b1; d2
sort:    b1; a2
sort:    b3; b1
sort:    b3; d2
sort:    c; b3
sort:    c; d2
filter:  a2
map:     a2
forEach: A2
filter:  b1
filter:  b3
filter:  c
filter:  d2

首先，排序操作在整個(gè)輸入集合上執(zhí)行。也就是說(shuō)，sorted以水平方式執(zhí)行。所以這里sorted對(duì)輸入集合中每個(gè)元素的多種組合調(diào)用了八次。

我們同樣可以通過(guò)重排調(diào)用鏈來(lái)優(yōu)化性能：

Stream.of("d2", "a2", "b1", "b3", "c")
    .filter(s -> {
        System.out.println("filter: " + s);
        return s.startsWith("a");
    })
    .sorted((s1, s2) -> {
        System.out.printf("sort: %s; %s
", s1, s2);
        return s1.compareTo(s2);
    })
    .map(s -> {
        System.out.println("map: " + s);
        return s.toUpperCase();
    })
    .forEach(s -> System.out.println("forEach: " + s));

// filter:  d2
// filter:  a2
// filter:  b1
// filter:  b3
// filter:  c
// map:     a2
// forEach: A2

這個(gè)例子中sorted永遠(yuǎn)不會(huì)調(diào)用，因?yàn)?b>filter把輸入集合減少至只有一個(gè)元素。所以對(duì)于更大的輸入集合會(huì)極大提升性能。

Java8的數(shù)據(jù)流不能被復(fù)用。一旦你調(diào)用了任何終止操作，數(shù)據(jù)流就關(guān)閉了：

Stream stream =
    Stream.of("d2", "a2", "b1", "b3", "c")
        .filter(s -> s.startsWith("a"));

stream.anyMatch(s -> true);    // ok
stream.noneMatch(s -> true);   // exception

在相同數(shù)據(jù)流上，在anyMatch之后調(diào)用noneMatch會(huì)產(chǎn)生下面的異常：

java.lang.IllegalStateException: stream has already been operated upon or closed
    at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:229)
    at java.util.stream.ReferencePipeline.noneMatch(ReferencePipeline.java:459)
    at com.winterbe.java8.Streams5.test7(Streams5.java:38)
    at com.winterbe.java8.Streams5.main(Streams5.java:28)

要克服這個(gè)限制，我們需要為每個(gè)我們想要執(zhí)行的終止操作創(chuàng)建新的數(shù)據(jù)流調(diào)用鏈。例如，我們創(chuàng)建一個(gè)數(shù)據(jù)流供應(yīng)器，來(lái)構(gòu)建新的數(shù)據(jù)流，并且設(shè)置好所有銜接操作：

Supplier> streamSupplier =
    () -> Stream.of("d2", "a2", "b1", "b3", "c")
            .filter(s -> s.startsWith("a"));

streamSupplier.get().anyMatch(s -> true);   // ok
streamSupplier.get().noneMatch(s -> true);  // ok

每次對(duì)get()的調(diào)用都構(gòu)造了一個(gè)新的數(shù)據(jù)流，我們將其保存來(lái)調(diào)用終止操作。

數(shù)據(jù)流執(zhí)行大量的不同操作。我們已經(jīng)了解了一些最重要的操作，例如filter和map。我將它們留給你來(lái)探索所有其他的可用操作（請(qǐng)見數(shù)據(jù)流的Javadoc）。下面讓我們深入了解一些更復(fù)雜的操作：collect、flatMap和reduce。

這一節(jié)的大部分代碼示例使用下面的Person列表來(lái)演示：

class Person {
    String name;
    int age;

    Person(String name, int age) {
        this.name = name;
        this.age = age;
    }

    @Override
    public String toString() {
        return name;
    }
}

List persons =
    Arrays.asList(
        new Person("Max", 18),
        new Person("Peter", 23),
        new Person("Pamela", 23),
        new Person("David", 12));

collect是非常有用的終止操作，將流中的元素存放在不同類型的結(jié)果中，例如List、Set或者Map。collect接受收集器（Collector），它由四個(gè)不同的操作組成：供應(yīng)器（supplier）、累加器（accumulator）、組合器（combiner）和終止器（finisher）。這在開始聽起來(lái)十分復(fù)雜，但是Java8通過(guò)內(nèi)置的Collectors類支持多種內(nèi)置的收集器。所以對(duì)于大部分常見操作，你并不需要自己實(shí)現(xiàn)收集器。

讓我們以一個(gè)非常常見的用例來(lái)開始：

List filtered =
    persons
        .stream()
        .filter(p -> p.name.startsWith("P"))
        .collect(Collectors.toList());

System.out.println(filtered);    // [Peter, Pamela]

就像你看到的那樣，它非常簡(jiǎn)單，只是從流的元素中構(gòu)造了一個(gè)列表。如果需要以Set來(lái)替代List，只需要使用Collectors.toSet()就好了。

下面的例子按照年齡對(duì)所有人進(jìn)行分組：

Map> personsByAge = persons
    .stream()
    .collect(Collectors.groupingBy(p -> p.age));

personsByAge
    .forEach((age, p) -> System.out.format("age %s: %s
", age, p));

// age 18: [Max]
// age 23: [Peter, Pamela]
// age 12: [David]

收集器十分靈活。你也可以在流的元素上執(zhí)行聚合，例如，計(jì)算所有人的平均年齡：

Double averageAge = persons
    .stream()
    .collect(Collectors.averagingInt(p -> p.age));

System.out.println(averageAge);     // 19.0

如果你對(duì)更多統(tǒng)計(jì)學(xué)方法感興趣，概要收集器返回一個(gè)特殊的內(nèi)置概要統(tǒng)計(jì)對(duì)象，所以我們可以簡(jiǎn)單計(jì)算最小年齡、最大年齡、算術(shù)平均年齡、總和和數(shù)量。

IntSummaryStatistics ageSummary =
    persons
        .stream()
        .collect(Collectors.summarizingInt(p -> p.age));

System.out.println(ageSummary);
// IntSummaryStatistics{count=4, sum=76, min=12, average=19.000000, max=23}

下面的例子將所有人連接為一個(gè)字符串：

String phrase = persons
    .stream()
    .filter(p -> p.age >= 18)
    .map(p -> p.name)
    .collect(Collectors.joining(" and ", "In Germany ", " are of legal age."));

System.out.println(phrase);
// In Germany Max and Peter and Pamela are of legal age.

連接收集器接受分隔符，以及可選的前綴和后綴。

為了將數(shù)據(jù)流中的元素轉(zhuǎn)換為映射，我們需要指定鍵和值如何被映射。要記住鍵必須是唯一的，否則會(huì)拋出IllegalStateException異常。你可以選擇傳遞一個(gè)合并函數(shù)作為額外的參數(shù)來(lái)避免這個(gè)異常。

既然我們知道了一些最強(qiáng)大的內(nèi)置收集器，讓我們來(lái)嘗試構(gòu)建自己的特殊收集器吧。我們希望將流中的所有人轉(zhuǎn)換為一個(gè)字符串，包含所有大寫的名稱，并以|分割。為了完成它，我們通過(guò)Collector.of()創(chuàng)建了一個(gè)新的收集器。我們需要傳遞一個(gè)收集器的四個(gè)組成部分：供應(yīng)器、累加器、組合器和終止器。

Collector personNameCollector =
    Collector.of(
        () -> new StringJoiner(" | "),          // supplier
        (j, p) -> j.add(p.name.toUpperCase()),  // accumulator
        (j1, j2) -> j1.merge(j2),               // combiner
        StringJoiner::toString);                // finisher

String names = persons
    .stream()
    .collect(personNameCollector);

System.out.println(names);  // MAX | PETER | PAMELA | DAVID

由于Java中的字符串是不可變的，我們需要一個(gè)助手類StringJointer。讓收集器構(gòu)造我們的字符串。供應(yīng)器最開始使用相應(yīng)的分隔符構(gòu)造了這樣一個(gè)StringJointer。累加器用于將每個(gè)人的大寫名稱加到StringJointer中。組合器知道如何把兩個(gè)StringJointer合并為一個(gè)。最后一步，終結(jié)器從StringJointer構(gòu)造出預(yù)期的字符串。

我們已經(jīng)了解了如何通過(guò)使用map操作，將流中的對(duì)象轉(zhuǎn)換為另一種類型。map有時(shí)十分受限，因?yàn)槊總€(gè)對(duì)象只能映射為一個(gè)其它對(duì)象。但如何我希望將一個(gè)對(duì)象轉(zhuǎn)換為多個(gè)或零個(gè)其他對(duì)象呢？flatMap這時(shí)就會(huì)派上用場(chǎng)。

flatMap將流中的每個(gè)元素，轉(zhuǎn)換為其它對(duì)象的流。所以每個(gè)對(duì)象會(huì)被轉(zhuǎn)換為零個(gè)、一個(gè)或多個(gè)其它對(duì)象，以流的形式返回。這些流的內(nèi)容之后會(huì)放進(jìn)flatMap所返回的流中。

在我們了解flatMap如何使用之前，我們需要相應(yīng)的類型體系：

class Foo {
    String name;
    List bars = new ArrayList<>();

    Foo(String name) {
        this.name = name;
    }
}

class Bar {
    String name;

    Bar(String name) {
        this.name = name;
    }
}

下面，我們使用我們自己的關(guān)于流的知識(shí)來(lái)實(shí)例化一些對(duì)象：

List foos = new ArrayList<>();

// create foos
IntStream
    .range(1, 4)
    .forEach(i -> foos.add(new Foo("Foo" + i)));

// create bars
foos.forEach(f ->
    IntStream
        .range(1, 4)
        .forEach(i -> f.bars.add(new Bar("Bar" + i + " <- " + f.name))));

現(xiàn)在我們擁有了含有三個(gè)foo的列表，每個(gè)都含有三個(gè)bar。

flatMap接受返回對(duì)象流的函數(shù)。所以為了處理每個(gè)foo上的bar對(duì)象，我們需要傳遞相應(yīng)的函數(shù)：

foos.stream()
    .flatMap(f -> f.bars.stream())
    .forEach(b -> System.out.println(b.name));

// Bar1 <- Foo1
// Bar2 <- Foo1
// Bar3 <- Foo1
// Bar1 <- Foo2
// Bar2 <- Foo2
// Bar3 <- Foo2
// Bar1 <- Foo3
// Bar2 <- Foo3
// Bar3 <- Foo3

像你看到的那樣，我們成功地將含有三個(gè)foo對(duì)象中的流轉(zhuǎn)換為含有九個(gè)bar對(duì)象的流。

最后，上面的代碼示例可以簡(jiǎn)化為流式操作的單一流水線：

IntStream.range(1, 4)
    .mapToObj(i -> new Foo("Foo" + i))
    .peek(f -> IntStream.range(1, 4)
        .mapToObj(i -> new Bar("Bar" + i + " <- " f.name))
        .forEach(f.bars::add))
    .flatMap(f -> f.bars.stream())
    .forEach(b -> System.out.println(b.name));

flatMap也可用于Java8引入的Optional類。Optional的flatMap操作返回一個(gè)Optional或其他類型的對(duì)象。所以它可以用于避免煩人的null檢查。

考慮像這樣更復(fù)雜的層次結(jié)構(gòu)：

class Outer {
    Nested nested;
}

class Nested {
    Inner inner;
}

class Inner {
    String foo;
}

為了處理外層示例上的內(nèi)層字符串foo，你需要添加多個(gè)null檢查來(lái)避免潛在的NullPointerException：

Outer outer = new Outer();
if (outer != null && outer.nested != null && outer.nested.inner != null) {
    System.out.println(outer.nested.inner.foo);
}

可以使用Optional的flatMap操作來(lái)完成相同的行為：

Optional.of(new Outer())
    .flatMap(o -> Optional.ofNullable(o.nested))
    .flatMap(n -> Optional.ofNullable(n.inner))
    .flatMap(i -> Optional.ofNullable(i.foo))
    .ifPresent(System.out::println);

如果存在的話，每個(gè)flatMap的調(diào)用都會(huì)返回預(yù)期對(duì)象的Optional包裝，否則為null的Optional包裝。

歸約操作將所有流中的元素組合為單一結(jié)果。Java8支持三種不同類型的reduce方法。第一種將流中的元素歸約為流中的一個(gè)元素。讓我們看看我們?nèi)绾问褂眠@個(gè)方法來(lái)計(jì)算出最老的人：

persons
    .stream()
    .reduce((p1, p2) -> p1.age > p2.age ? p1 : p2)
    .ifPresent(System.out::println);    // Pamela

reduce方法接受BinaryOperator積累函數(shù)。它實(shí)際上是兩個(gè)操作數(shù)類型相同的BiFunction。BiFunction就像是Function，但是接受兩個(gè)參數(shù)。示例中的函數(shù)比較兩個(gè)人的年齡，來(lái)返回年齡較大的人。

第二個(gè)reduce方法接受一個(gè)初始值，和一個(gè)BinaryOperator累加器。這個(gè)方法可以用于從流中的其它Person對(duì)象中構(gòu)造帶有聚合后名稱和年齡的新Person對(duì)象。

Person result =
    persons
        .stream()
        .reduce(new Person("", 0), (p1, p2) -> {
            p1.age += p2.age;
            p1.name += p2.name;
            return p1;
        });

System.out.format("name=%s; age=%s", result.name, result.age);
// name=MaxPeterPamelaDavid; age=76

第三個(gè)reduce對(duì)象接受三個(gè)參數(shù)：初始值，BiFunction累加器和BinaryOperator類型的組合器函數(shù)。由于初始值的類型不一定為Person，我們可以使用這個(gè)歸約函數(shù)來(lái)計(jì)算所有人的年齡總和。：

Integer ageSum = persons
    .stream()
    .reduce(0, (sum, p) -> sum += p.age, (sum1, sum2) -> sum1 + sum2);

System.out.println(ageSum);  // 76

你可以看到結(jié)果是76。但是背后發(fā)生了什么？讓我們通過(guò)添加一些調(diào)試輸出來(lái)擴(kuò)展上面的代碼：

Integer ageSum = persons
    .stream()
    .reduce(0,
        (sum, p) -> {
            System.out.format("accumulator: sum=%s; person=%s
", sum, p);
            return sum += p.age;
        },
        (sum1, sum2) -> {
            System.out.format("combiner: sum1=%s; sum2=%s
", sum1, sum2);
            return sum1 + sum2;
        });

// accumulator: sum=0; person=Max
// accumulator: sum=18; person=Peter
// accumulator: sum=41; person=Pamela
// accumulator: sum=64; person=David

你可以看到，累加器函數(shù)做了所有工作。它首先使用初始值0和第一個(gè)人Max來(lái)調(diào)用累加器。接下來(lái)的三步中sum會(huì)持續(xù)增加，直到76。

等一下。好像組合器從來(lái)沒(méi)有調(diào)用過(guò)？以并行方式執(zhí)行相同的流會(huì)揭開這個(gè)秘密：

Integer ageSum = persons
    .parallelStream()
    .reduce(0,
        (sum, p) -> {
            System.out.format("accumulator: sum=%s; person=%s
", sum, p);
            return sum += p.age;
        },
        (sum1, sum2) -> {
            System.out.format("combiner: sum1=%s; sum2=%s
", sum1, sum2);
            return sum1 + sum2;
        });

// accumulator: sum=0; person=Pamela
// accumulator: sum=0; person=David
// accumulator: sum=0; person=Max
// accumulator: sum=0; person=Peter
// combiner: sum1=18; sum2=23
// combiner: sum1=23; sum2=12
// combiner: sum1=41; sum2=35

這個(gè)流的并行執(zhí)行行為會(huì)完全不同。現(xiàn)在實(shí)際上調(diào)用了組合器。由于累加器被并行調(diào)用，組合器需要用于計(jì)算部分累加值的總和。

下一節(jié)我們會(huì)深入了解并行流。

流可以并行執(zhí)行，在大量輸入元素上可以提升運(yùn)行時(shí)的性能。并行流使用公共的ForkJoinPool，由ForkJoinPool.commonPool()方法提供。底層線程池的大小最大為五個(gè)線程 -- 取決于CPU的物理核數(shù)。

ForkJoinPool commonPool = ForkJoinPool.commonPool();
System.out.println(commonPool.getParallelism());    // 3

在我的機(jī)器上，公共池默認(rèn)初始化為3。這個(gè)值可以通過(guò)設(shè)置下列JVM參數(shù)來(lái)增減：

-Djava.util.concurrent.ForkJoinPool.common.parallelism=5

集合支持parallelStream()方法來(lái)創(chuàng)建元素的并行流。或者你可以在已存在的數(shù)據(jù)流上調(diào)用銜接方法parallel()，將串行流轉(zhuǎn)換為并行流。

為了描述并行流的執(zhí)行行為，下面的例子向sout打印了當(dāng)前線程的信息。

Arrays.asList("a1", "a2", "b1", "c2", "c1")
    .parallelStream()
    .filter(s -> {
        System.out.format("filter: %s [%s]
",
            s, Thread.currentThread().getName());
        return true;
    })
    .map(s -> {
        System.out.format("map: %s [%s]
",
            s, Thread.currentThread().getName());
        return s.toUpperCase();
    })
    .forEach(s -> System.out.format("forEach: %s [%s]
",
        s, Thread.currentThread().getName()));

通過(guò)分析調(diào)試輸出，我們可以對(duì)哪個(gè)線程用于執(zhí)行流式操作擁有更深入的理解：

filter:  b1 [main]
filter:  a2 [ForkJoinPool.commonPool-worker-1]
map:     a2 [ForkJoinPool.commonPool-worker-1]
filter:  c2 [ForkJoinPool.commonPool-worker-3]
map:     c2 [ForkJoinPool.commonPool-worker-3]
filter:  c1 [ForkJoinPool.commonPool-worker-2]
map:     c1 [ForkJoinPool.commonPool-worker-2]
forEach: C2 [ForkJoinPool.commonPool-worker-3]
forEach: A2 [ForkJoinPool.commonPool-worker-1]
map:     b1 [main]
forEach: B1 [main]
filter:  a1 [ForkJoinPool.commonPool-worker-3]
map:     a1 [ForkJoinPool.commonPool-worker-3]
forEach: A1 [ForkJoinPool.commonPool-worker-3]
forEach: C1 [ForkJoinPool.commonPool-worker-2]

就像你看到的那樣，并行流使用了所有公共的ForkJoinPool中的可用線程來(lái)執(zhí)行流式操作。在連續(xù)的運(yùn)行中輸出可能有所不同，因?yàn)樗褂玫奶囟ň€程是非特定的。

讓我們通過(guò)添加額外的流式操作sort來(lái)擴(kuò)展這個(gè)示例：

Arrays.asList("a1", "a2", "b1", "c2", "c1")
    .parallelStream()
    .filter(s -> {
        System.out.format("filter: %s [%s]
",
            s, Thread.currentThread().getName());
        return true;
    })
    .map(s -> {
        System.out.format("map: %s [%s]
",
            s, Thread.currentThread().getName());
        return s.toUpperCase();
    })
    .sorted((s1, s2) -> {
        System.out.format("sort: %s <> %s [%s]
",
            s1, s2, Thread.currentThread().getName());
        return s1.compareTo(s2);
    })
    .forEach(s -> System.out.format("forEach: %s [%s]
",
        s, Thread.currentThread().getName()));

結(jié)果起初可能比較奇怪：

filter:  c2 [ForkJoinPool.commonPool-worker-3]
filter:  c1 [ForkJoinPool.commonPool-worker-2]
map:     c1 [ForkJoinPool.commonPool-worker-2]
filter:  a2 [ForkJoinPool.commonPool-worker-1]
map:     a2 [ForkJoinPool.commonPool-worker-1]
filter:  b1 [main]
map:     b1 [main]
filter:  a1 [ForkJoinPool.commonPool-worker-2]
map:     a1 [ForkJoinPool.commonPool-worker-2]
map:     c2 [ForkJoinPool.commonPool-worker-3]
sort:    A2 <> A1 [main]
sort:    B1 <> A2 [main]
sort:    C2 <> B1 [main]
sort:    C1 <> C2 [main]
sort:    C1 <> B1 [main]
sort:    C1 <> C2 [main]
forEach: A1 [ForkJoinPool.commonPool-worker-1]
forEach: C2 [ForkJoinPool.commonPool-worker-3]
forEach: B1 [main]
forEach: A2 [ForkJoinPool.commonPool-worker-2]
forEach: C1 [ForkJoinPool.commonPool-worker-1]

sort看起來(lái)只在主線程上串行執(zhí)行。實(shí)際上，并行流上的sort在背后使用了Java8中新的方法Arrays.parallelSort()。如javadoc所說(shuō)，這個(gè)方法會(huì)參照數(shù)據(jù)長(zhǎng)度來(lái)決定以串行或并行來(lái)執(zhí)行。

如果指定數(shù)據(jù)的長(zhǎng)度小于最小粒度，它使用相應(yīng)的Arrays.sort方法來(lái)排序。

返回上一節(jié)中reduce的例子。我們已經(jīng)發(fā)現(xiàn)了組合器函數(shù)只在并行流中調(diào)用，而不在串行流中調(diào)用。讓我們來(lái)觀察實(shí)際上涉及到哪個(gè)線程：

List persons = Arrays.asList(
    new Person("Max", 18),
    new Person("Peter", 23),
    new Person("Pamela", 23),
    new Person("David", 12));

persons
    .parallelStream()
    .reduce(0,
        (sum, p) -> {
            System.out.format("accumulator: sum=%s; person=%s [%s]
",
                sum, p, Thread.currentThread().getName());
            return sum += p.age;
        },
        (sum1, sum2) -> {
            System.out.format("combiner: sum1=%s; sum2=%s [%s]
",
                sum1, sum2, Thread.currentThread().getName());
            return sum1 + sum2;
        });

控制臺(tái)的輸出表明，累加器和組合器都在所有可用的線程上并行執(zhí)行：

accumulator: sum=0; person=Pamela; [main]
accumulator: sum=0; person=Max;    [ForkJoinPool.commonPool-worker-3]
accumulator: sum=0; person=David;  [ForkJoinPool.commonPool-worker-2]
accumulator: sum=0; person=Peter;  [ForkJoinPool.commonPool-worker-1]
combiner:    sum1=18; sum2=23;     [ForkJoinPool.commonPool-worker-1]
combiner:    sum1=23; sum2=12;     [ForkJoinPool.commonPool-worker-2]
combiner:    sum1=41; sum2=35;     [ForkJoinPool.commonPool-worker-2]

總之，并行流對(duì)擁有大量輸入元素的數(shù)據(jù)流具有極大的性能提升。但是要記住一些并行流的操作，例如reduce和collect需要額外的計(jì)算（組合操作），這在串行執(zhí)行時(shí)并不需要。

此外我們已經(jīng)了解，所有并行流操作都共享相同的JVM相關(guān)的公共ForkJoinPool。所以你可能需要避免實(shí)現(xiàn)又慢又卡的流式操作，因?yàn)樗赡軙?huì)拖慢你應(yīng)用中嚴(yán)重依賴并行流的其它部分。

我的Java8數(shù)據(jù)流編程教程就此告一段落。如果你對(duì)深入了解Java8數(shù)據(jù)流感興趣，我向你推薦數(shù)據(jù)流的Javadoc。如果你希望學(xué)到更多底層機(jī)制，你可能需要閱讀Martin Fowler關(guān)于集合流水線的文章。

如果你對(duì)JavaScript也感興趣，你可能希望看一看Stream.js -- 一個(gè)Java8數(shù)據(jù)流API的JavaScript實(shí)現(xiàn)。你也可能希望閱讀我的Java8簡(jiǎn)明教程，和我的Java8Nashron教程。

我希望你會(huì)喜歡這篇文章。如果你有任何的問(wèn)題都可以在下面評(píng)論或者通過(guò) Twitter 給我回復(fù)。

祝編程愉快！