Java8 并行流(parallelStream)原理分析及线程池线数设置

Bertha 。 2023-01-18 09:17 67阅读 0赞

我们都知道在java 使用strem流做多线程处理是非常方便的。

list.parallelStream().forEach(s -> { 
                // 业务处理
            });

但是parallelStream是如何实现多线程处理的呢？其实看源码我们会发现parallelStream是使用线程池ForkJoin来调度的，这里我们可以看源码看到  
由于我们使用的 forEach，所以直接看类`ForEachOps`即可.当然其他方法比如reduce等会有不同的实现类，总的实现类是如下几个

java.util.stream.FindOps.FindOp
    java.util.stream.ForEachOps
    java.util.stream.MatchOps.MatchOp
    java.util.stream.ReduceOps.ReduceOp

里面基本都是调用 `evaluateParallel`方法  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjUxOTA0_size_16_color_FFFFFF_t_70]  
而 ForEachTask 继承关系如下  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjUxOTA0_size_16_color_FFFFFF_t_70 1]  
这里就找到了我们的答案，核心原理就是Fork Join 框架  
而ForkJoinPool的默认线程数是CPU核数 - 1，无参构造方法就可以看出  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjUxOTA0_size_16_color_FFFFFF_t_70 2]

如何设置 ForkJoinPool的线程数呢  
这里有如下几种方法

1.  直接构建自己的 ForkJoinPool

CountDownLatch countDownLatch = new CountDownLatch(20);
            int cpu = Runtime.getRuntime().availableProcessors();
            System.out.println(cpu);
            ForkJoinPool pool = new ForkJoinPool(2);
            List<Integer> list = IntStream.range(0, 20).boxed().collect(Collectors.toList());
            pool.submit(() -> { 
                list.parallelStream().forEach(s -> { 
                    // 业务处理
                    System.out.println("thread:" + Thread.currentThread().getName() + "value" + s);
                    countDownLatch.countDown();
                });
            });
            countDownLatch.await();

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjUxOTA0_size_16_color_FFFFFF_t_70 3]

可以看到始终只有2两个线程

*  使用参数设置

System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "5");

demo验证

System.setProperty(
                    "java.util.concurrent.ForkJoinPool.common.parallelism", "5");
            CountDownLatch countDownLatch = new CountDownLatch(5);
            List<Integer> list = IntStream.range(0, 20).boxed().collect(Collectors.toList());
            list.parallelStream().forEach(s -> { 
                // 业务处理
                System.out.println("thread:" + Thread.currentThread().getName() + "value" + s);
                countDownLatch.countDown();
    
            });
            countDownLatch.await();
            int poolSize = ForkJoinPool.commonPool().getPoolSize();
            System.out.println("Pool size: " + poolSize);

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjUxOTA0_size_16_color_FFFFFF_t_70 4]  
这里比较奇怪是设置了5个线程，虽然只有5个，但是主线程也加入了执行，这里暂时不做详细研究，有知道的欢迎留言

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjUxOTA0_size_16_color_FFFFFF_t_70]: /images/20221021/f1b90950b2f94c99a5a9cffe86180389.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjUxOTA0_size_16_color_FFFFFF_t_70 1]: /images/20221021/46ce71889348400eb8a45d3eedf1bd72.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjUxOTA0_size_16_color_FFFFFF_t_70 2]: /images/20221021/5c8031806d624beda024b34d83156a30.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjUxOTA0_size_16_color_FFFFFF_t_70 3]: /images/20221021/c0b18fb2ad234698b280a26189f32047.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjUxOTA0_size_16_color_FFFFFF_t_70 4]: /images/20221021/01f07f0201e244209599fcaea19f603b.png