深入浅出Java lambda的parallelStream
原文：https://blog.csdn.net/u011001723/article/details/52794455
什么是Stream？
Stream是java8中新增加的一个特性,被java猿统称为流.
Stream 不是集合元素，它不是资料结构并不储存资料，它是有关算法和计算的，它更像一个高阶版本的 Iterator。原始版本的 Iterator，使用者只能显式地一个一个遍历元素并对其执行某些操作；高阶版本的 Stream，使用者只要给出需要对其包含的元素执行什么操作，比如 "过滤掉长度大于 10 的字串"、"获取每个字串的首字母"等，Stream 会隐式地在内部进行遍历，做出相应的资料转换。
Stream 就如同一个迭代器（Iterator），单向，不可往复，资料只能遍历一次，遍历过一次后即用尽了，就好比流水从面前流过，一去不复返。
而和迭代器又不同的是，Stream 可以并行化操作，迭代器只能命令式地、序列化操作。顾名思义，当使用序列方式去遍历时，每个 item 读完后再读下一个 item。而使用并行去遍历时，资料会被分成多个段，其中每一个都在不同的执行绪中处理，然后将结果一起输出。Stream 的并行操作依赖于 Java7 中引入的 Fork/Join 框架（JSR166y）来拆分任务和加速处理过程。Java 的并行 API 演变历程基本如下：
1.0-1.4 中的 java.lang.Thread
5.0 中的 java.util.concurrent
6.0 中的 Phasers 等
7.0 中的 Fork/Join 框架
8.0 中的 Lambda
Stream 的另外一大特点是，资料来源本身可以是无限的。
parallelStream是什么
parallelStream其实就是一个并行执行的流.它通过预设的ForkJoinPool,可能提高你的多执行绪任务的速度.
parallelStream的作用
Stream具有平行处理能力，处理的过程会分而治之，也就是将一个大任务切分成多个小任务，这表示每个任务都是一个操作，因此像以下的程式片段：
List numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9);
numbers.parallelStream()
.forEach(out::println);
你得到的展示顺序不一定会是1、2、3、4、5、6、7、8、9，而可能是任意的顺序，就forEach()这个操作来讲，如果平行处理时，希望最后顺序是按照原来Stream的资料顺序，那可以呼叫forEachOrdered()。例如：
List numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9);
numbers.parallelStream()
.forEachOrdered(out::println);
注意: 如果forEachOrdered()中间有其他如filter()的中介操作，会试着平行化处理，然后最终forEachOrdered()会以原资料顺序处理，因此，使用forEachOrdered()这类的有序处理,可能会（或完全失去）失去平行化的一些优势，实际上中介操作亦有可能如此，例如sorted()方法。
parallelStream背后的男人:ForkJoinPool
要想深入的研究parallelStream之前,那么我们必须先了解ForkJoin框架和ForkJoinPool.本文旨在parallelStream,但因为两种关系甚密,故在此简单介绍一下ForkJoinPool,如有兴趣可以更深入的去了解下ForkJoin***(当然,如果你想真正的搞透parallelStream,那么你依然需要先搞透ForkJoinPool). *
ForkJoin框架是从jdk7中新特性,它同ThreadPoolExecutor一样，也实现了Executor和ExecutorService界面。它使用了一个无限伫列来储存需要执行的任务，而执行绪的数量则是通过建构函式传入，如果没有向建构函式中传入希望的执行绪数量，那么当前计算机可用的CPU数量会被设定为执行绪数量作为预设值。
ForkJoinPool主要用来使用 分治法(Divide-and-Conquer Algorithm) 来解决问题。典型的应用比如快速排序算法。这里的要点在于，ForkJoinPool需要使用相对少的执行绪来处理大量的任务。比如要对1000万个资料进行排序，那么会将这个任务分割成两个500万的排序任务和一个针对这两组500万资料的合并任务。以此类推，对于500万的资料也会做出同样的分割处理，到最后会设定一个阈值来规定当资料规模到多少时，停止这样的分割处理。比如，当元素的数量小于10时，会停止分割，转而使用插入排序对它们进行排序。那么到最后，所有的任务加起来会有大概2000000+个。问题的关键在于，对于一个任务而言，只有当它所有的子任务完成之后，它才能够被执行。
所以当使用ThreadPoolExecutor时，使用分治法会存在问题，因为ThreadPoolExecutor中的执行绪无法像任务伫列中再新增一个任务并且在等待该任务完成之后再继续执行。而使用ForkJoinPool时，就能够让其中的执行绪建立新的任务，并挂起当前的任务，此时执行绪就能够从伫列中选择子任务执行。
那么使用ThreadPoolExecutor或者ForkJoinPool，会有什么效能的差异呢？ 首先，使用ForkJoinPool能够使用数量有限的执行绪来完成非常多的具有父子关系的任务，比如使用4个执行绪来完成超过200万个任务。但是，使用ThreadPoolExecutor时，是不可能完成的，因为ThreadPoolExecutor中的Thread无法选择优先执行子任务，需要完成200万个具有父子关系的任务时，也需要200万个执行绪，显然这是不可行的。
工作窃取算法
forkjoin最核心的地方就是利用了现代硬件装置多核,在一个操作时候会有空闲的cpu,那么如何利用好这个空闲的cpu就成了提高效能的关键,而这里我们要提到的工作窃取（work-stealing）算法就是整个forkjion框架的核心理念,工作窃取（work-stealing）算法是指某个执行绪从其他伫列里窃取任务来执行。
那么为什么需要使用工作窃取算法呢？ 假如我们需要做一个比较大的任务，我们可以把这个任务分割为若干互不依赖的子任务，为了减少执行绪间的竞争，于是把这些子任务分别放到不同的伫列里，并为每个伫列建立一个单独的执行绪来执行伫列里的任务，执行绪和伫列一一对应，比如A执行绪负责处理A伫列里的任务。但是有的执行绪会先把自己伫列里的任务干完，而其他执行绪对应的伫列里还有任务等待处理。干完活的执行绪与其等著，不如去帮其他执行绪干活，于是它就去其他执行绪的伫列里窃取一个任务来执行。而在这时它们会访问同一个伫列，所以为了减少窃取任务执行绪和被窃取任务执行绪之间的竞争，通常会使用双端伫列，被窃取任务执行绪永远从双端伫列的头部拿任务执行，而窃取任务的执行绪永远从双端伫列的尾部拿任务执行。
工作窃取算法的优点是充分利用执行绪进行平行计算，并减少了执行绪间的竞争，其缺点是在某些情况下还是存在竞争，比如双端伫列里只有一个任务时。并且消耗了更多的系统资源，比如建立多个执行绪和多个双端伫列。
用看forkJoin的眼光来看ParallelStreams
上文中已经提到了在Java 8引入了自动并行化的概念。它能够让一部分Java程式码自动地以并行的方式执行，也就是我们使用了ForkJoinPool的ParallelStream。
Java 8为ForkJoinPool添加了一个通用执行绪池，这个执行绪池用来处理那些没有被显式提交到任何执行绪池的任务。它是ForkJoinPool型别上的一个静态元素，它拥有的预设执行绪数量等于执行计算机上的处理器数量。当呼叫Arrays类上新增的新方法时，自动并行化就会发生。比如用来排序一个数组的并行快速排序，用来对一个数组中的元素进行并行遍历。自动并行化也被运用在Java 8新新增的Stream API中。
比如下面的程式码用来遍历列表中的元素并执行需要的操作：
List userInfoList =
DaoContainers.getUserInfoDAO().queryAllByList(new UserInfoModel());
userInfoList.parallelStream().forEach(RedisUserApi::setUserIdUserInfo);
对于列表中的元素的操作都会以并行的方式执行。forEach方法会为每个元素的计算操作建立一个任务，该任务会被前文中提到的ForkJoinPool中的通用执行绪池处理。以上的平行计算逻辑当然也可以使用ThreadPoolExecutor完成，但是就程式码的可读性和程式码量而言，使用ForkJoinPool明显更胜一筹。
对于ForkJoinPool通用执行绪池的执行绪数量，通常使用预设值就可以了，即执行时计算机的处理器数量。我这里提供了一个示例的程式码让你了解jvm所使用的ForkJoinPool的执行绪数量, 你可以可以通过设定系统属性：-Djava.util.concurrent.ForkJoinPool.common.parallelism=N （N为执行绪数量）,来调整ForkJoinPool的执行绪数量,可以尝试调整成不同的引数来观察每次的输出结果:
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.CountDownLatch;
/** * @description 这是一个用来让你更加熟悉parallelStream的原理的实力 * @date 2016年10月11日18:26:55 * @version v1.0 * @author wangguangdong */
public class App {
public static void main(String[] args) throws Exception {
System.out.println("Hello World!");
// 构造一个10000个元素的集合
List list = new ArrayList();
for (int i = 0; i list.add(i);
}
// 统计并行执行list的执行绪
Set threadSet = new CopyOnWriteArraySet();
// 并行执行
list.parallelStream().forEach(integer -> {
Thread thread = Thread.currentThread();
// System.out.println(thread);
// 统计并行执行list的执行绪
threadSet.add(thread);
});
System.out.println("threadSet一共有" + threadSet.size() + "个执行绪");
System.out.println("系统一个有"+Runtime.getRuntime().availableProcessors()+"个cpu");
List list1 = new ArrayList();
List list2 = new ArrayList();
for (int i = 0; i list1.add(i);
list2.add(i);
}
Set threadSetTwo = new CopyOnWriteArraySet();
CountDownLatch countDownLatch = new CountDownLatch(2);
Thread threadA = new Thread(() -> {
list1.parallelStream().forEach(integer -> {
Thread thread = Thread.currentThread();
// System.out.println("list1" + thread);
threadSetTwo.add(thread);
});
countDownLatch.countDown();
});
Thread threadB = new Thread(() -> {
list2.parallelStream().forEach(integer -> {
Thread thread = Thread.currentThread();
// System.out.println("list2" + thread);
threadSetTwo.add(thread);
});
countDownLatch.countDown();
});
threadA.start();
threadB.start();
countDownLatch.await();
System.out.print("threadSetTwo一共有" + threadSetTwo.size() + "个执行绪");
System.out.println("---------------------------");
System.out.println(threadSet);
System.out.println(threadSetTwo);
System.out.println("---------------------------");
threadSetTwo.addAll(threadSet);
System.out.println(threadSetTwo);
System.out.println("threadSetTwo一共有" + threadSetTwo.size() + "个执行绪");
System.out.println("系统一个有"+Runtime.getRuntime().availableProcessors()+"个cpu");
}
}
出现这种现象的原因是，forEach方法用了一些小把戏。它会将执行forEach本身的执行绪也作为执行绪池中的一个工作执行绪。因此，即使将ForkJoinPool的通用执行绪池的执行绪数量设定为1，实际上也会有2个工作执行绪。因此在使用forEach的时候，执行绪数为1的ForkJoinPool通用执行绪池和执行绪数为2的ThreadPoolExecutor是等价的。
所以当ForkJoinPool通用执行绪池实际需要4个工作执行绪时，可以将它设定成3，那么在执行时可用的工作执行绪就是4了。
小结:
1. 当需要处理递回分治算法时，考虑使用ForkJoinPool。
2. 仔细设定不再进行任务划分的阈值，这个阈值对效能有影响。
3. Java 8中的一些特性会使用到ForkJoinPool中的通用执行绪池。在某些场合下，需要调整该执行绪池的预设的执行绪数量。
ParallelStreams 的陷阱
上文中我们已经看到了ParallelStream他强大无比的特性,但这里我们就讲告诉你ParallelStreams不是万金油,而是一把双刃剑,如果错误的使用反倒可能伤人伤己.
以下是一个我们专案里使用 parallel streams 的很常见的情况。在这个例子中，我们想同时呼叫不同地址的api中并且获得第一个返回的结果。
public static String query(String q, List engines) {
Optional result = engines.stream().parallel().map((base) -> {
String url = base + q;
return WS.url(url).get();
}).findAny();
return result.get();
}
可能有很多朋友在jdk7用future配合countDownLatch自己实现的这个功能,但是jdk8的朋友基本都会用上面的实现方式,那么自信深究一下究竟自己用future实现的这个功能和利用jdk8的parallelStream来实现这个功能有什么 不同点 呢? 坑又在哪里呢 ？
让我们细思思考一下整个功能究竟是如何运转的。首先我们的集合元素engines 由ParallelStreams并行的去进行map操作 (ParallelStreams使用JVM预设的forkJoin框架的执行绪池由当前执行绪去执行并行操作).
然而,这里需要注意的一地方是我们在呼叫第三方的api请求是一个响应略慢而且会阻塞操作的一个过程。所以在某时刻所有执行绪都会呼叫 get() 方法并且在那里等待结果返回.
再回过头仔细思考一下这个功能的实现过程是我们一开始想要的吗？我们是在同一时间等待所有的结果,而不是遍历这个列表按顺序等待每个回答.然而，由于ForkJoinPool workders的存在，这样平行的等待相对于使用主执行绪的等待会产生的一种副作用.
现在 ForkJoin pool (关于forkjion的更多实现你可以去搜索引擎中去看一下他的具体实现方式) 的实现是: 它并不会因为产生了新的workers而抵消掉阻塞的workers。那么在某个时间所有 ForkJoinPool.common() 的执行绪都会被用光.也就是说，下一次你呼叫这个查询方法，就可能会在一个时间与其他的parallel stream同时执行，而导致第二个任务的效能大大受损。或者说，例如你在这个功能里是用来快速返回呼叫的第三方api的,而在其他的功能里是用于一些简单的资料平行计算的,但是假如你先呼叫了这个功能,同一时间之后呼叫计算的函式,那么这里forkjionPool的实现会让你计算的函式大打折扣.
不过也不要急着去吐槽ForkJoinPool的实现,在不同的情况下你可以给它一个ManagedBlocker例项并且确保它知道在一个阻塞呼叫中应该什么时候去抵消掉卡住的workers.现在有意思的一点是，在一个parallel stream处理中并不一定是阻塞呼叫会拖延程式的效能。任何被用于对映在一个集合上的长时间执行的函式都会产生同样的问题.
正如我们上面那个列子的情况分析得知,lambda的执行并不是瞬间完成的,所有使用parallel streams的程式都有可能成为阻塞程式的源头,并且在执行过程中程式中的其他部分将无法访问这些workers,这意味着任何依赖parallel streams的程式在什么别的东西占用着common ForkJoinPool时将会变得不可预知并且暗藏危机.
怎么正确使用parallelStream
如果你正在写一个其他地方都是单执行绪的程式并且准确地知道什么时候你应该要使用parallel streams，这样的话你可能会觉得这个问题有一点肤浅。然而，我们很多人是在处理web应用、各种不同的框架以及重量级应用服务。一个服务器是怎样被设计成一个可以支援多种独立应用的主机的？谁知道呢，给你一个可以并行的却不能控制输入的parallel stream.
很抱歉,请原谅我用的标注 [怎么正确使用parallelStream] ,因为目前为止我也没有发现一个好的方式来让我真正的正确使用parallelStream.下面的网上写的两种方式:
一种方式是限制ForkJoinPool提供的并行数。可以通过使用-Djava.util.concurrent.ForkJoinPool.common.parallelism=1 来限制执行绪池的大小为1。不再从并行化中得到好处可以杜绝错误的使用它 (其实这个方式还是有点搞笑的,既然这样搞那我还不如不去使用并行流) 。
另一种方式就是，一个被称为工作区的可以让ForkJoinPool平行放置的 parallelStream() 实现。不幸的是现在的JDK还没有实现。
Parallel streams 是无法预测的，而且想要正确地使用它有些棘手。几乎任何parallel streams的使用都会影响程式中无关部分的效能，而且是一种无法预测的方式。。但是在呼叫stream.parallel() 或者parallelStream()时候在我的程式码里之前我仍然会重新审视一遍他给我的程式究竟会带来什么问题,他能有多大的提升,是否有使用他的意义.
stream or parallelStream？
上面我们也看到了parallelStream所带来的隐患和好处,那么,在从stream和parallelStream方法中进行选择时,我们可以考虑以下几个问题：
1. 是否需要并行？
2. 任务之间是否是独立的？是否会引起任何竞态条件？
3. 结果是否取决于任务的呼叫顺序？
对于问题1，在回答这个问题之前，你需要弄清楚你要解决的问题是什么，资料量有多大，计算的特点是什么？并不是所有的问题都适合使用并发程式来求解，比如当资料量不大时，顺序执行往往比并行执行更快。毕竟，准备执行绪池和其它相关资源也是需要时间的。但是，当任务涉及到I/O操作并且任务之间不互相依赖时，那么并行化就是一个不错的选择。通常而言，将这类程式并行化之后，执行速度会提升好几个等级。
对于问题2，如果任务之间是独立的，并且程式码中不涉及到对同一个物件的某个状态或者某个变数的更新操作，那么就表明程式码是可以被并行化的。
对于问题3，由于在并行环境中任务的执行顺序是不确定的，因此对于依赖于顺序的任务而言，并行化也许不能给出正确的结果。
