ParallelStream的坑，不踩不知道，一踩吓一跳

本文转载自微信公众号「小姐姐味道」，作者小姐姐养的狗  。转载本文请联系小姐姐味道公众号。 

很多同学喜欢使用lambda表达式，它允许你定义短小精悍的函数，体现你高超的编码水平。当然，这个功能在某些以代码行数来衡量工作量的公司来说，就比较吃亏一些。

比如下面的代码片段，让人阅读的时候就像是读诗一样。但是一旦用不好，也是会要命的。

 
 
 
    
  
  
  
  1. List transactionsIds = 
    2. 
  
  
  
  2. widgets.stream() 
    3. 
  
  
  
  3.              .filter(b -> b.getColor() == RED) 
    4. 
  
  
  
  4.              .sorted((x,y) -> x.getWeight() - y.getWeight()) 
    5. 
  
  
  
  5.              .mapToInt(Widget::getWeight) 
    6. 
  
  
  
  6.              .sum(); 
    7.

这段代码有一个关键的函数，那就是stream。通过它，可以将一个普通的list，转化为流，然后就可以使用类似于管道的方式对list进行操作。总之，用过的都说好。

对这些函数还不是太熟悉?可以参考：《到处是map、flatMap，啥意思?》

问题来了

假如我们把stream换成parallelStream，会发生什么情况?

根据字面上的意思，流会从串行 变成并行。

既然是并行，那用屁股想一想，就知道这里面肯定会有线程安全问题。不过我们这里讨论的并不是要你使用线程安全的集合，这个话题太低级。现阶段，知道在线程不安全的环境中使用线程安全的集合，已经是一个基本的技能。

这次踩坑的地方，是并行流的性能问题。

我们用代码来说话。

下面的代码，开启了8个线程，这8个线程都在使用并行流进行数据计算。在执行的逻辑中，我们让每个任务都sleep 1秒钟，这样就能够模拟一些I/O请求的耗时等待。

使用stream，程序会在30秒后返回，但我们期望程序能够在1秒多返回，因为它是并行流，得对得起这个称号。

测试发现，我们等了好久，任务才执行完毕。

 
 
 
    
  
  
  
  1. static void paralleTest() { 
    2. 
  
  
  
  2.     List numbers = Arrays.asList( 
    3. 
  
  
  
  3.             0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 
    4. 
  
  
  
  4.             10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 
    5. 
  
  
  
  5.             20, 21, 22, 23, 24, 25, 26, 27, 28, 29 
    6. 
  
  
  
  6.     ); 
    7. 
  
  
  
  7.     final long begin = System.currentTimeMillis(); 
    8. 
  
  
  
  8.     numbers.parallelStream().map(k -> { 
    9. 
  
  
  
  9.         try { 
    10. 
  
  
  
  10.             Thread.sleep(1000); 
    11. 
  
  
  
  11.             System.out.println((System.currentTimeMillis() - begin) + "ms => " + k + " \t" + Thread.currentThread()); 
    12. 
  
  
  
  12.         } catch (InterruptedException e) { 
    13. 
  
  
  
  13.             e.printStackTrace(); 
    14. 
  
  
  
  14.         } 
    15. 
  
  
  
  15.         return k; 
    16. 
  
  
  
  16.     }).collect(Collectors.toList()); 
    17. 
  
  
  
  17. } 
    18. 
  
  
  
  18.  
    19. 
  
  
  
  19. public static void main(String[] args) { 
    20. 
  
  
  
  20. //    System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "20"); 
    21. 
  
  
  
  21.     new Thread(() -> paralleTest()).start(); 
    22. 
  
  
  
  22.     new Thread(() -> paralleTest()).start(); 
    23. 
  
  
  
  23.     new Thread(() -> paralleTest()).start(); 
    24. 
  
  
  
  24.     new Thread(() -> paralleTest()).start(); 
    25. 
  
  
  
  25.     new Thread(() -> paralleTest()).start(); 
    26. 
  
  
  
  26.     new Thread(() -> paralleTest()).start(); 
    27. 
  
  
  
  27.     new Thread(() -> paralleTest()).start(); 
    28. 
  
  
  
  28.     new Thread(() -> paralleTest()).start(); 
    29. 
  
  
  
  29. } 
    30.

坑

实际上，在不同的机器上执行，这段代码花费的时间都不一样。

既然是并行，那肯定得有个并行度。太低了，体现不到并行的能能力;太大了，又浪费了上下文切换的时间。我是很沮丧的发现，很多高级研发，将线程池的各种参数背的滚瓜烂熟，各种调优，竟然敢睁一只眼闭一只眼的在I/O密集型业务中用上parallelStream。

要了解这个并行度，我们需要查看具体的构造方法。在ForkJoinPool类中找到这样的代码。

 
 
 
    
  
  
  
  1. try {  // ignore exceptions in accessing/parsing properties 
    2. 
  
  
  
  2.     String pp = System.getProperty 
    3. 
  
  
  
  3.         ("java.util.concurrent.ForkJoinPool.common.parallelism"); 
    4. 
  
  
  
  4.     if (pp != null) 
    5. 
  
  
  
  5.         parallelism = Integer.parseInt(pp); 
    6. 
  
  
  
  6.     fac = (ForkJoinWorkerThreadFactory) newInstanceFromSystemProperty( 
    7. 
  
  
  
  7.         "java.util.concurrent.ForkJoinPool.common.threadFactory"); 
    8. 
  
  
  
  8.     handler = (UncaughtExceptionHandler) newInstanceFromSystemProperty( 
    9. 
  
  
  
  9.         "java.util.concurrent.ForkJoinPool.common.exceptionHandler"); 
    10. 
  
  
  
  10. } catch (Exception ignore) { 
    11. 
  
  
  
  11. } 
    12. 
  
  
  
  12.  
    13. 
  
  
  
  13. if (fac == null) { 
    14. 
  
  
  
  14.     if (System.getSecurityManager() == null) 
    15. 
  
  
  
  15.         fac = defaultForkJoinWorkerThreadFactory; 
    16. 
  
  
  
  16.     else // use security-managed default 
    17. 
  
  
  
  17.         fac = new InnocuousForkJoinWorkerThreadFactory(); 
    18. 
  
  
  
  18. } 
    19. 
  
  
  
  19. if (parallelism < 0 && // default 1 less than #cores 
    20. 
  
  
  
  20.     (parallelism = Runtime.getRuntime().availableProcessors() - 1) <= 0) 
    21. 
  
  
  
  21.     parallelism = 1; 
    22. 
  
  
  
  22. if (parallelism > MAX_CAP) 
    23. 
  
  
  
  23.     parallelism = MAX_CAP; 
    24.

可以看到，并行度到底是多少，是由下面的参数来控制的。如果无法获取这个参数，则默认使用 CPU个数-1 的并行度。

可以看到，这个函数是为了计算密集型业务去设计的。如果你喂给它一大堆任务，它就会由并行执行退变成类似于串行的效果。

 
 
 
    
  
  
  
  1. -Djava.util.concurrent.ForkJoinPool.common.parallelism=N 
    2.

即使你使用-Djava.util.concurrent.ForkJoinPool.common.parallelism=N设置了一个初始值大小，它依然有问题。

因为，parallelism这个变量是final的，一旦设定，不允许修改。也就是说，上面的参数只会生效一次。

张三可能使用下面的代码，设置了并行度大小为20。

 
 
 
    
  
  
  
  1. System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "20"); 
    2.

李四可能用同样的方式，设置了这个值为30。那实际在项目中用的是哪个值，那就得问JVM是怎么加载的类信息了。

这种方式并不太非常靠谱。

一种解决方式

我们可以通过提供外置的forkjoinpool，也就是改变提交方式，来实现不同类型的任务分离。

代码如下所示，通过显式的代码提交，即可实现任务分离。

 
 
 
    
  
  
  
  1. ForkJoinPool pool = new ForkJoinPool(30); 
    2. 
  
  
  
  2.  
    3. 
  
  
  
  3. final long begin = System.currentTimeMillis(); 
    4. 
  
  
  
  4. try { 
    5. 
  
  
  
  5.     pool.submit(() -> 
    6. 
  
  
  
  6.             numbers.parallelStream().map(k -> { 
    7. 
  
  
  
  7.                 try { 
    8. 
  
  
  
  8.                     Thread.sleep(1000); 
    9. 
  
  
  
  9.                     System.out.println((System.currentTimeMillis() - begin) + "ms => " + k + " \t" + Thread.currentThread()); 
    10. 
  
  
  
  10.                 } catch (InterruptedException e) { 
    11. 
  
  
  
  11.                     e.printStackTrace(); 
    12. 
  
  
  
  12.                 } 
    13. 
  
  
  
  13.                 return k; 
    14. 
  
  
  
  14.             }).collect(Collectors.toList())).get(); 
    15. 
  
  
  
  15. } catch (InterruptedException e) { 
    16. 
  
  
  
  16.     e.printStackTrace(); 
    17. 
  
  
  
  17. } catch (ExecutionException e) { 
    18. 
  
  
  
  18.     e.printStackTrace(); 
    19. 
  
  
  
  19. } 
    20.

这样，不同的场景，就可以拥有不同的并行度。这种方式和CountDownLatch有异曲同工之妙，我们需要手动管理资源。

使用了这种方式，代码量增加，已经和优雅关系不大了，不仅不优雅，而且丑的要命。白天鹅变成了丑小鸭，你还会爱它么?

作者简介：小姐姐味道 (xjjdog)，一个不允许程序员走弯路的公众号。聚焦基础架构和Linux。十年架构，日百亿流量，与你探讨高并发世界，给你不一样的味道。我的个人微信xjjdog0，欢迎添加好友，进一步交流。

网页题目：ParallelStream的坑，不踩不知道，一踩吓一跳
网站地址：http://www.shufengxianlan.com/qtweb/news1/424801.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联

猜你还喜欢下面的内容

手机网站建设知识

同城分类信息

网站服务器租用　　 　网站建设公司　　 　做网站　　 　护栏钻孔机　　 　腾讯云虚拟主机　　 　移动APP　　 　网络推广平台　　 　新都网站建设　　 　德阳东方电机技改　　 　定制网站建设　　 　php空间　　 　软文营销　　 　达州网站设计　　 　艾名斯智能　　 　成都二枢机房　　 　精灵网　　 　企业网站建设　　 　四川雕琢时光　　 　微信公众号开发　　 　和正兴机械