摘要：實(shí)際上，在并行流上使用新的方法。此外，我們了解到所有并行流操作共享相同的范圍。因此，您可能希望避免實(shí)施慢速阻塞流操作，因?yàn)檫@可能會(huì)減慢嚴(yán)重依賴并行流的應(yīng)用程序的其他部分。

流可以并行執(zhí)行，以增加大量輸入元素的運(yùn)行時(shí)性能。并行流ForkJoinPool通過(guò)靜態(tài)ForkJoinPool.commonPool()方法使用公共可用的流。底層線程池的大小最多使用五個(gè)線程 - 具體取決于可用物理CPU核心的數(shù)量：

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

在我的機(jī)器上，公共池初始化為默認(rèn)值為3的并行度。通過(guò)設(shè)置以下JVM參數(shù)可以減小或增加此值：

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

集合支持創(chuàng)建并行元素流的方法parallelStream()?；蛘?，您可以在給定流上調(diào)用中間方法parallel()，以將順序流轉(zhuǎn)換為并行流。

為了評(píng)估并行流的并行執(zhí)行行為，下一個(gè)示例將有關(guān)當(dāng)前線程的信息打印出來(lái)：

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)查調(diào)試輸出，我們應(yīng)該更好地理解哪些線程實(shí)際用于執(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]

如您所見(jiàn)，并行流利用公共中的所有可用線程ForkJoinPool來(lái)執(zhí)行流操作。輸出在連續(xù)運(yùn)行中可能不同，因?yàn)閷?shí)際使用的特定線程的行為是非確定性的。

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

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é)果可能最初看起來(lái)很奇怪：

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í)行。實(shí)際上，sort在并行流上使用新的Java 8方法Arrays.parallelSort()。如Javadoc中所述，如果排序?qū)错樞蚧虿⑿袌?zhí)行，則此方法決定數(shù)組的長(zhǎng)度：

如果指定數(shù)組的長(zhǎng)度小于最小粒度，則使用適當(dāng)?shù)腁rrays.sort方法對(duì)其進(jìn)行排序。

回到reduce一節(jié)的例子。我們已經(jīng)發(fā)現(xiàn)組合器函數(shù)只是并行調(diào)用，而不是順序流調(diào)用。讓我們看看實(shí)際涉及哪些線程：

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)輸出顯示累加器和組合器函數(shù)在所有可用線程上并行執(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]

總之，并行流可以為具有大量輸入元素的流帶來(lái)良好的性能提升。但請(qǐng)記住，某些并行流操作reduce,collect需要額外的計(jì)算（組合操作），這在順序執(zhí)行時(shí)是不需要的。

此外，我們了解到所有并行流操作共享相同的JVM范圍ForkJoinPool。因此，您可能希望避免實(shí)施慢速阻塞流操作，因?yàn)檫@可能會(huì)減慢嚴(yán)重依賴并行流的應(yīng)用程序的其他部分。