Java中parallelStream的使用

本文主要是介绍Java中parallelStream的使用，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

前言: 各位读者们好久不见，最近博主我一直在实习中，而且也准备校招，刷了很多很多的题目，今天在实习中看到了师兄写的代码中使用到了parallelStream()这个并行流计算，所以我想写一篇博客总结一下这个parallelStream新特性的具体使用。

1、什么是parallelStream?

众所周知，stream流式计算是Java8中新添的一种新特性，可以把stream流理解为串行的流式计算，而parallelStream是一种多线程并行流式计算。

2、parallelStream的作用在哪？或者说有什么好处?

在stream()中任务的调度执行是串行化的，需要执行完了其中一个任务执行下一个，而parallelStream是利用多线程进行的，这可以很大程度简化我们使用并发操作。

3、parallelStream怎么用?

先来看看stream的使用

        List<String> str = new ArrayList<>();str.add("1");str.add("2");str.add("3");str.add("4");str.add("5");str.add("6");/*单个线程执行任务*/str.stream().filter(e -> {System.out.println(Thread.currentThread().getName() + "\t过滤" + e);return Integer.parseInt(e) % 2 == 0 ? true : false;}).collect(Collectors.toList());

执行结果
在这里插入图片描述
可以看到所有的任务都是一个线程完成的

再来看看 parallelStream

      List<String> str = new ArrayList<>();str.add("1");str.add("2");str.add("3");str.add("4");str.add("5");str.add("6");str.parallelStream().filter(e -> {System.out.println(Thread.currentThread().getName() + "\t过滤" + e);return Integer.parseInt(e) % 2 == 0 ? true : false;}).collect(Collectors.toList());

执行结果:
在这里插入图片描述
可以看到任务是并行执行的并且有些事使用了worker线程。

此时小伙伴们可能疑惑了，出现ForkJoinPool这是什么东西，其实parallelStream底层就是使用了Fork/Join 分支计算最终合并的技术，具体的原理小编会在后面的博客中去讲解，这里就不具体的展开了

4、parallelStream的注意点或者缺点?

在展开具体说这个的时候，我们先来看一个demo

    public void streamTest(){List<Integer> list = Arrays.asList(1,2,3,4,5,6,7,8);list.parallelStream().forEach(e -> {System.out.println("第一次并行" + Thread.currentThread().getName() + "\t执行" + e);try {TimeUnit.SECONDS.sleep(5L);} catch (InterruptedException interruptedException) {interruptedException.printStackTrace();}});}public void streamTest2(){List<Integer> list = Arrays.asList(1,2,3,4,5,6,7,8);list.parallelStream().forEach(e -> {System.out.println("第二次并行" + Thread.currentThread().getName() + "\t执行" + e);try {TimeUnit.SECONDS.sleep(5L);} catch (InterruptedException interruptedException) {interruptedException.printStackTrace();}});}@Testpublic void test3(){new Thread(() ->{streamTest();},"AA").start();new Thread(() ->{streamTest2();},"BB").start();while (Thread.activeCount() > 2){}}

执行结果：
在这里插入图片描述
可以看得出来所有的第二次并行的fork/join线程执行之后，才执行第一次并行放入fork/join 这时候整体就变成了串行执行。这时候会引发如下的一些问题

在虚拟机启动时，我们指定了worker线程的数量

Djava.util.concurrent.ForkJoinPool.common.parallelism=N

整个程序的生命周期都将使用这些工作线程, 这必然存在任务生产和消费的问题，如果某个生产者生产了许多重量级的任务（耗时很长），那么其他任务毫无疑问将会没有工作线程可用；更可怕的事情是这些工作线程正在进行IO阻塞。本应利用并行加速处理的业务，因为工作者不够反而会额外增加处理时间，使得系统性能在某一时刻大打折扣。而且这一类问题往往是很难排查的。我们并不知道一个重量级项目中的哪一个框架、哪一个模块在使用并行流。