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

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

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

在函數(shù)式編程中，單體是一個結(jié)構(gòu)，表示定義為步驟序列的計算。單體結(jié)構(gòu)的類型定義了它對鏈式操作，或具有相同類型的嵌套函數(shù)的含義。

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

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

更新 - 我現(xiàn)在正在編寫用于瀏覽器的Java8數(shù)據(jù)流API的JavaScript實現(xiàn)。如果你對此感興趣，請在Github上訪問Stream.js。非常期待你的反饋。

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

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ù)流操作要么是銜接操作，要么是終止操作。銜接操作返回數(shù)據(jù)流，所以我們可以把多個銜接操作不使用分號來鏈接到一起。終止操作無返回值，或者返回一個不是流的結(jié)果。在上面的例子中，filter、map和sorted都是銜接操作，而forEach是終止操作。列表上的所有流式操作請見數(shù)據(jù)流的Javadoc。你在上面例子中看到的這種數(shù)據(jù)流的鏈式操作也叫作操作流水線。

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

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

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

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

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

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

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

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

除了普通的對象數(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ù)流都像通常的對象數(shù)據(jù)流一樣，但有一些不同。基本的數(shù)據(jù)流使用特殊的lambda表達式，例如，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ù)據(jù)流轉(zhuǎn)換為基本數(shù)據(jù)流，或者相反。出于這種目的，對象數(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ù)流可以通過maoToObj()轉(zhuǎn)換為對象數(shù)據(jù)流：

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

// a1
// a2
// a3

下面是組合示例：浮點數(shù)據(jù)流首先映射為整數(shù)數(shù)據(jù)流，之后映射為字符串的對象數(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í)行吧。

銜接操作的一個重要特性就是延遲性。觀察下面沒有終止操作的例子：

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

執(zhí)行這段代碼時，不向控制臺打印任何東西。這是因為銜接操作只在終止操作調(diào)用時被執(zhí)行。

讓我們通過添加終止操作forEach來擴展這個例子：

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

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

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

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

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

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操作就會返回true。對于第二個傳遞"A2"的元素，它的結(jié)果為真。由于數(shù)據(jù)流的鏈式調(diào)用是垂直執(zhí)行的，map這里只需要執(zhí)行兩次。所以map會執(zhí)行盡可能少的次數(shù)，而不是把所有元素都映射一遍。

下面的例子由兩個銜接操作map和filter，以及一個終止操作forEach組成。讓我們再來看看這些操作如何執(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會對底層集合的每個字符串調(diào)用五次，而forEach只會調(diào)用一次。

如果我們調(diào)整操作順序，將filter移動到調(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只會調(diào)用一次，所以操作流水線對于更多的輸入元素會執(zhí)行更快。在整合復(fù)雜的方法鏈時，要記住這一點。

讓我們通過添加額外的方法sorted來擴展上面的例子：

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)的操作，因為你需要在處理中保存狀態(tài)來對集合中的元素排序。

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

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

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

我們同樣可以通過重排調(diào)用鏈來優(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

這個例子中sorted永遠不會調(diào)用，因為filter把輸入集合減少至只有一個元素。所以對于更大的輸入集合會極大提升性能。

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會產(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)

要克服這個限制，我們需要為每個我們想要執(zhí)行的終止操作創(chuàng)建新的數(shù)據(jù)流調(diào)用鏈。例如，我們創(chuàng)建一個數(shù)據(jù)流供應(yīng)器，來構(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

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

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

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

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），它由四個不同的操作組成：供應(yīng)器（supplier）、累加器（accumulator）、組合器（combiner）和終止器（finisher）。這在開始聽起來十分復(fù)雜，但是Java8通過內(nèi)置的Collectors類支持多種內(nèi)置的收集器。所以對于大部分常見操作，你并不需要自己實現(xiàn)收集器。

讓我們以一個非常常見的用例來開始：

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

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

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

下面的例子按照年齡對所有人進行分組：

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í)行聚合，例如，計算所有人的平均年齡：

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

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

如果你對更多統(tǒng)計學(xué)方法感興趣，概要收集器返回一個特殊的內(nèi)置概要統(tǒng)計對象，所以我們可以簡單計算最小年齡、最大年齡、算術(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}

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

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)換為映射，我們需要指定鍵和值如何被映射。要記住鍵必須是唯一的，否則會拋出IllegalStateException異常。你可以選擇傳遞一個合并函數(shù)作為額外的參數(shù)來避免這個異常。

既然我們知道了一些最強大的內(nèi)置收集器，讓我們來嘗試構(gòu)建自己的特殊收集器吧。我們希望將流中的所有人轉(zhuǎn)換為一個字符串，包含所有大寫的名稱，并以|分割。為了完成它，我們通過Collector.of()創(chuàng)建了一個新的收集器。我們需要傳遞一個收集器的四個組成部分：供應(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中的字符串是不可變的，我們需要一個助手類StringJointer。讓收集器構(gòu)造我們的字符串。供應(yīng)器最開始使用相應(yīng)的分隔符構(gòu)造了這樣一個StringJointer。累加器用于將每個人的大寫名稱加到StringJointer中。組合器知道如何把兩個StringJointer合并為一個。最后一步，終結(jié)器從StringJointer構(gòu)造出預(yù)期的字符串。

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

flatMap將流中的每個元素，轉(zhuǎn)換為其它對象的流。所以每個對象會被轉(zhuǎn)換為零個、一個或多個其它對象，以流的形式返回。這些流的內(nèi)容之后會放進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)于流的知識來實例化一些對象：

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)在我們擁有了含有三個foo的列表，每個都含有三個bar。

flatMap接受返回對象流的函數(shù)。所以為了處理每個foo上的bar對象，我們需要傳遞相應(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

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

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

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操作返回一個Optional或其他類型的對象。所以它可以用于避免煩人的null檢查。

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

class Outer {
    Nested nested;
}

class Nested {
    Inner inner;
}

class Inner {
    String foo;
}

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

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

可以使用Optional的flatMap操作來完成相同的行為：

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);

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

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

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

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

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

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

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

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

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

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

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和第一個人Max來調(diào)用累加器。接下來的三步中sum會持續(xù)增加，直到76。

等一下。好像組合器從來沒有調(diào)用過？以并行方式執(zhí)行相同的流會揭開這個秘密：

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

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

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

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

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

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

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

集合支持parallelStream()方法來創(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()));

通過分析調(diào)試輸出，我們可以對哪個線程用于執(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中的可用線程來執(zhí)行流式操作。在連續(xù)的運行中輸出可能有所不同，因為所使用的特定線程是非特定的。

讓我們通過添加額外的流式操作sort來擴展這個示例：

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看起來只在主線程上串行執(zhí)行。實際上，并行流上的sort在背后使用了Java8中新的方法Arrays.parallelSort()。如javadoc所說，這個方法會參照數(shù)據(jù)長度來決定以串行或并行來執(zhí)行。

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

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

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

控制臺的輸出表明，累加器和組合器都在所有可用的線程上并行執(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]

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

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

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

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

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

祝編程愉快！