Collection

一、方法概览

推荐一篇博客 我也是参考 水很深

二、什么是流Stream（此流非彼流）

• Stream是java8新增特性，被人称之为流
• Stream不是集合元素，它不是数据结构，因此它不保存数据，它是有关算法和计算的，它更像一个高级版本的Iterator.原始版本的Iterator只能显示的一个一个遍历元素并对其执行某种操作，而Stream只需要给出需要对其包含的元素执行什么操作，比如“过滤长度为10的字符串”等 Stream会隐式的在内部进行遍历，做出相应的数据操作。
• Stream与迭代器一样是单向的，不可重复的，数据只遍历一次
• Stream可以进行并行化操作，迭代器只能命令式的、串行化操作
• 当时用串行化方式去遍历时，每个item读完后才能读取下一个item。而是用并行去遍历时，数据会被分割成多个段，其中每一个段都在不同的线程中处理，然后将结果一起输出。
• Stream的并行操作依赖于java7引入的Fork/Join框架来拆分任务处理过程。
• JAVA并行API演变历程
  1）1.0-1.4 java.lang.Thread
  2 ) 1.5 java.util.concurrent
  3 ) 1.6 Phasers 等
  4） 1.7 Fork/Join 框架
  5）1.8 Lambda
• Stream的另一个大特点是，数据源本身可以是无限的

二、parallelStream

1. 什么是 parallelStream
   parallelStream 是一个并行执行的流，他通过默认的ForkJoinPool ，可能提高多线程任务的速度
2. parallelStream作用
   Stream具有平行处理的能力，处理的过程会分而治之，也就是将一个大任务切分成多个小任务（跟hadoop mapreduce一个意思吧），这表示每个任务都是一个操作
   
   输出结果~~：
   

• 可以看到 展示不是按照顺序来的，
• 当然forEach 改成forEachOrdered 结果会是顺序的，
• 如果forEachOrdered 中间有其他 如filter()的中介操作，他会试着平行化处理，最终forEachOrdered 会以原数据的顺序处理，因此forEachOrdered 处理可能会失去平行化的优势

三、 ForkJoin

1. ForkJoinPool是什么
   想深入研究parallelStream 我们必须先了解ForkJoin框架和ForkJonPool ,因为parallelStream 依赖于ForkJonPool ，所以这里做简单介绍
   ForkJoin框架是jdk1.7的新特性，他与ThreadPoolExecutor一样，也实现了Executor和ExecutorService接口。它使用一个无限的队列保存需要执行的任务，而线程的数量则是通过构造函数传入，如果没有构造函数中传入数据，那么就会用当前计算机可用CPU数量作为线程的默认值
   2.ForkJoinPool作用

• ForkJoinPool主要使用分治法来解决问题。典型的应用比如快速排序算法。ForkjoinPool需要使用相对少的线程来处理大量的任务（比如10个线程 执行10000个任务）
  – ForkJoinPool 能让其中的线程创建新的任务，并挂起当前任务，此时线程就能够在队列中选择任务执行（ThreadPoolExecutor就不行）
• 也就是说使用ForkJoinPool能够使用数量有限的线程来完成非常多的具有父子关系的任务，比如使用10个线程来完成100万任务，但是，使用ThreadPoolExeutor是不可能完成的，因为ThreadPoolExecutor中的Thread无法选择优先执行子任务，需要完成100万个具有父子任务时，业务要100万线程，这显然不合理~

四、窃取算法

1. 窃取算法是什么？
   ForkJoin最核心的地方就是利用现代硬件设备多核，在一个操作的时候会有空闲的cpu，那么如何利用好这个空闲的cpu就成了提高性能的关键，而这里我们要提到工作窃取算法，工作窃取算法是整个ForkJoin的核心，工作窃取算法值某个线程从其他队列里窃取任务来执行。
2. 为什么需要使用窃取算法？
   假如我们需要做一个比较大的任务，我们可以把这个任务分割为若干互不依赖的子任务，为了减少线程间的竞争，于是把这些子任务分别放到不同的队列里，并为每个队列创建一个独立的线程来执行队列里的任务，线程队列一一对应，比如X线程负责处理X队列里的任务。但是有的线程会先把自己手里的任务完成，而其他线程对应的队列里的任务等待处理。完成任务的线程与其等着，不如去帮其他线程做任务，于是它就去其他队里里边窃取一个任务来执行，而在这时它们会同时访问同一个队列，所以为了减少窃取任务线程和被窃取任务线程之间的竞争，通常 会使用双端队列，被窃取的线程永远从双端队列的头部拿任务执行，而窃取任务的线程永远从双端队列的尾部拿任务执行。

六、ForkJion ParallelStreams

Java8给ForkJoinPool添加了一个通用的线程池，这个线程池用来处理那些没有被显式提交到任何线程池的任务。他是ForkJoinPool类型上的一个静态元素，他拥有默认线程数量。当调用Arrays类上添加的新方法时（parallelSort ），自动并行化就会发生。
对于ForkJoinPool通用线程池的线程数量，通常使用默认值就可以了，即运行时计算机的数量。

package along;import java.util.ArrayList;import java.util.List;import java.util.Set;import java.util.concurrent.CopyOnWriteArraySet;import java.util.concurrent.CountDownLatch;public class javaTest {public static void main(String[] args) throws Exception {
		System.out.println("Hello World!");// 构造一个10000个元素的集合
		List<Integer> list = new ArrayList<>();for (int i = 0; i < 10000; i++) {
			list.add(i);}// 统计并行执行list的线程
		Set<Thread> 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<Integer> list1 = new ArrayList<>();
		List<Integer> list2 = new ArrayList<>();for (int i = 0; i < 100000; i++) {
			list1.add(i);
			list2.add(i);}
		Set<Thread> threadSetTwo = new CopyOnWriteArraySet<>();
		CountDownLatch countDownLatch = new CountDownLatch(2);
		Thread threadA = new Thread(() -> {
			list1.parallelStream().forEach(integer -> {
				Thread thread = Thread.currentThread();
				threadSetTwo.add(thread);});
			countDownLatch.countDown();});
		Thread threadB = new Thread(() -> {
			list2.parallelStream().forEach(integer -> {
				Thread thread = Thread.currentThread();
				threadSetTwo.add(thread);});
			countDownLatch.countDown();});

		threadA.start();
		threadB.start();
		countDownLatch.await();
		System.out.print("threadSetTwo一共有" + threadSetTwo.size() + "个线程");}}//threadSet是 4 也就是是3个线程池线程+1个forEach线程本身//threadSetTwo 是5 也就是3个线程池线程+2个forEach线程//所以如果ForkJoinPool需要4个线程 将它的值设置为 3 即可

ParallelStreams 的陷阱
parallelStreams不是万能的 不可乱用!

在这个例子中，我们想同时查询多个搜索引擎并且获得第一个返回的结果。

• 标红处，有问题！！
• 查询搜索引擎是一个阻塞操作。所以在某时刻所有线程都会调用get()方法并且在那里等待结果返回
• 由于ForkJoinPool workders的存在，这样平行的等待相对于使用主线程的等待会产生的一种副作用。
• 现在ForkJoin pool 的实现是:它并不会因为产生了新的workers 而抵消掉阻塞的workers。那么在某个时间所有ForkJoinPool.common()的线程都会被用光。
• 也就是说，下一次你调用这个查询方法，就可能会在一个时间与其他的parallel stream同时运行（因为还有上边的get()没有执行完），而导致第二个任务的性能大大受损。

小结：

1. 当需要处理递归分治算法时，考虑使用ForkJoinPool。
2. 仔细设置不再进行任务划分的阈值，这个阈值对性能有影响。
3. Java 8中的一些特性会使用到ForkJoinPool中的通用线程池。在某些场合下，需要调整该线程池的默认的线程数量。