你的位置:首页 > Java教程

[Java教程]Java 8并行流:必备技巧


本文由 ImportNew - paddx 翻译自 tobyhobson。欢迎加入翻译小组。转载请见文末要求。

Java 8 并行流(parallel stream)采用共享线程池,对性能造成了严重影响。可以包装流来调用自己的线程池解决性能问题。

问题

Java 8 的并行流可以让我们相对轻松地执行并行任务。

1
myList.parallelStream.map(obj -> longRunningOperation())



但是这样存在一个严重的问题:在 JVM 的后台,使用通用的 fork/join 池来完成上述功能,该池是所有并行流共享的。默认情况,fork/join 池会为每个处理器分配一个线程。假设你有一台16核的机器,这样你就只能创建16个线程。对 CPU 密集型的任务来说,这样是有意义的,因为你的机器确实只能执行16个线程。但是真实情况下,不是所有的任务都是 CPU 密集型的。例如:

1
2
3
4
5
6
7
8
9
myList.parallelStream 
   .map(this::retrieveFromA)
   .map(this::processUsingB)
   .forEach(this::saveToC)
 
myList.parallelStream 
   .map(this::retrieveFromD)
   .map(this::processUsingE)
   .forEach(this::saveToD)



这两个流很大程度上是受限于IO操作,所以会等待其他系统。但这两个流使用相同的(小)线程池,因此会相互等待而被阻塞。这个非常不好,可以改进。我们以一个流为例:

1
2
3
4
5
6
7
final List<Integer> firstRange = buildIntRange(); 
   firstRange.parallelStream().forEach((number) -> {
      try {
         // do something slow
         Thread.sleep(5);
      } catch (InterruptedException e) { }
});



完整的代码可以在gist上查看。

在执行期间,我获取了一份线程dump的文件。这是相关的线程(在我的Macbook上):

1
2
3
4
ForkJoinPool.commonPool-worker-1 
ForkJoinPool.commonPool-worker-2 
ForkJoinPool.commonPool-worker-3 
ForkJoinPool.commonPool-worker-4



现在,我要并行的执行这两个并行流(对于那些不是以英语为母语的人士,我感到非常抱歉!)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
Runnable firstTask = () -> { 
   firstRange.parallelStream().forEach((number) -> {
      try {
         // do something slow
         Thread.sleep(5);
      } catch (InterruptedException e) { }
   });
};
 
Runnable secondTask = () -> { 
   secondRange.parallelStream().forEach((number) -> {
      try {
         // do something slow
         Thread.sleep(5);
      } catch (InterruptedException e) { }
   });
};
// run threads



完整的代码可以在gist上查看。

这次我们再看一下线程dump文件:

1
2
3
4
ForkJoinPool.commonPool-worker-1 
ForkJoinPool.commonPool-worker-2 
ForkJoinPool.commonPool-worker-3 
ForkJoinPool.commonPool-worker-4



正如你所见,结果是一样的。我们只使用了4个线程。

一种变通方案

正如我所提到的,JVM 后台使用 fork/join 池,在 ForkJoinTask 的文档中,我们可以看到:

如果合适,安排一个异步执行的任务到当前正在运行的池中。如果任务不在inForkJoinPool()中,也可以调用ForkJoinPool.commonPool()获取新的池来执行。

让我试一试……

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
ForkJoinPool forkJoinPool = new ForkJoinPool(3); 
forkJoinPool.submit(() -> { 
    firstRange.parallelStream().forEach((number) -> {
        try {
            Thread.sleep(5);
        } catch (InterruptedException e) { }
    });
});
 
ForkJoinPool forkJoinPool2 = new ForkJoinPool(3); 
forkJoinPool2.submit(() -> { 
    secondRange.parallelStream().forEach((number) -> {
        try {
            Thread.sleep(5);
        } catch (InterruptedException e) {
        }
    });
});



完整的代码可以在gist上查看。

现在,我们再次查看线程池:

1
2
3
4
5
6
7
8
ForkJoinPool-1-worker-1 
ForkJoinPool-1-worker-2 
ForkJoinPool-1-worker-3 
ForkJoinPool-1-worker-4 
ForkJoinPool-2-worker-1 
ForkJoinPool-2-worker-2 
ForkJoinPool-2-worker-3 
ForkJoinPool-1-worker-4



因为我们创建自己的线程池,所以可以避免共享线程池,如果有需要,甚至可以分配比处理机数量更多的线程。

1
ForkJoinPool forkJoinPool = new ForkJoinPool(<numThreads>);



原文链接: tobyhobson 翻译: ImportNew.com - paddx
译文链接: http://www.importnew.com/16801.html
转载请保留原文出处、译者和译文链接。]

关于作者: paddx


 
全能程序员交流QQ群290551701,群内程序员都是来自,百度、阿里、京东、小米、去哪儿、饿了吗、蓝港等高级程序员 ,拥有丰富的经验。加入我们,直线沟通技术大牛,最佳的学习环境,了解业内的一手的资讯。如果你想结实大牛,那 就加入进来,让大牛带你超神!