博客
关于我
强烈建议你试试无所不能的chatGPT,快点击我
Java 8 Stream并行流
阅读量:5960 次
发布时间:2019-06-19

本文共 4461 字,大约阅读时间需要 14 分钟。

流可以并行执行,以增加大量输入元素的运行时性能。并行流ForkJoinPool通过静态ForkJoinPool.commonPool()方法使用公共可用的流。底层线程池的大小最多使用五个线程 - 具体取决于可用物理CPU核心的数量:

ForkJoinPool commonPool = ForkJoinPool.commonPool();System.out.println(commonPool.getParallelism()); // 3复制代码

在我的机器上,公共池初始化为默认值为3的并行度。通过设置以下JVM参数可以减小或增加此值:

-Djava.util.concurrent.ForkJoinPool.common.parallelism=5复制代码

集合支持创建并行元素流的方法parallelStream()。或者,您可以在给定流上调用中间方法parallel(),以将顺序流转换为并行流。

为了评估并行流的并行执行行为,下一个示例将有关当前线程的信息打印出来:

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

通过调查调试输出,我们应该更好地理解哪些线程实际用于执行流操作:

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来执行流操作。输出在连续运行中可能不同,因为实际使用的特定线程的行为是非确定性的。

让我们通过一个额外的流操作来扩展该示例:

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

结果可能最初看起来很奇怪:

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只在主线程上顺序执行。实际上,sort在并行流上使用新的Java 8方法Arrays.parallelSort()。如中所述,如果排序将按顺序或并行执行,则此方法决定数组的长度:

如果指定数组的长度小于最小粒度,则使用适当的Arrays.sort方法对其进行排序。

回到一节的例子。我们已经发现组合器函数只是并行调用,而不是顺序流调用。让我们看看实际涉及哪些线程:

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]\n", sum, p, Thread.currentThread().getName()); return sum += p.age; }, (sum1, sum2) -> { System.out.format("combiner: sum1=%s; sum2=%s [%s]\n", sum1, sum2, Thread.currentThread().getName()); return sum1 + sum2; });复制代码

控制台输出显示累加器和组合器函数在所有可用线程上并行执行:

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]复制代码

总之,并行流可以为具有大量输入元素的流带来良好的性能提升。但请记住,某些并行流操作reduce,collect需要额外的计算(组合操作),这在顺序执行时是不需要的。

此外,我们了解到所有并行流操作共享相同的JVM范围ForkJoinPool。因此,您可能希望避免实施慢速阻塞流操作,因为这可能会减慢严重依赖并行流的应用程序的其他部分。

转载地址:http://dguax.baihongyu.com/

你可能感兴趣的文章