java8 stram并行流学习_综合

关于steam基本用法https://www.jianshu.com/p/11c925cdba50这篇博客中已经有了详细的介绍

并行流需要注意的地方：

并行流分解任务基于fork/join 框架

fork/join框架是jdk1.7引入的，java8的stream多线程并非流的正是以这个框架为基础的，所以想要深入理解并发流就要学习fork/join框架。
fork/join框架的目的是以递归方式将可以并行的任务拆分成更小的任务，然后将每个子任务的结果合并起来生成整体结果。它是ExecutorService接口的一个实现，它把子任务分配线程池（ForkJoinPool）中的工作线程。要把任务提交到这个线程池，必须创建RecursiveTask的一个子类，如果任务不返回结果则是RecursiveAction的子类。

fork/join框架流程示意图：在这里插入图片描述

mport java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveTask;
import java.util.stream.LongStream;/*** Created by sunjin on 2016/7/5.* 继承RecursiveTask来创建可以用于分支/合并的框架任务*/
public class ForkJoinSumCalculator extends RecursiveTask<Long> {
    //要求和的数组private final long[] numbers;//子任务处理的数组开始和终止的位置private final int start;private final int end;//不在将任务分解成子任务的阀值大小public static final int THRESHOLD = 10000;//用于创建组任务的构造函数public ForkJoinSumCalculator(long[] numbers){
    this(numbers, 0, numbers.length);}//用于递归创建子任务的构造函数public ForkJoinSumCalculator(long[] numbers,int start,int end){
    this.numbers = numbers;this.start = start;this.end = end;}//重写接口的方法@Overrideprotected Long compute() {
    //当前任务负责求和的部分的大小int length = end - start;//如果小于等于阀值就顺序执行计算结果if(length <= THRESHOLD){
    return computeSequentially();}//创建子任务来为数组的前一半求和ForkJoinSumCalculator leftTask = new ForkJoinSumCalculator(numbers, start, start + length/2);//将子任务拆分出去，丢到ForkJoinPool线程池异步执行。leftTask.fork();//创建子任务来为数组的后一半求和ForkJoinSumCalculator rightTask = new ForkJoinSumCalculator(numbers, start + length/2, end);//第二个任务直接使用当前线程计算而不再开启新的线程。long rightResult = rightTask.compute();//读取第一个子任务的结果，如果没有完成则等待。long leftResult = leftTask.join();//合并两个子任务的计算结果return rightResult + leftResult;}//顺序执行计算的简单算法private long computeSequentially(){
    long sum = 0;for(int i =start; i< end; i++){
    sum += numbers[i];}return sum;}//提供给外部使用的入口方法public static long forkJoinSum(long n) {
    long[] numbers = LongStream.rangeClosed(1, n).toArray();ForkJoinTask<Long> task = new ForkJoinSumCalculator(numbers);return new ForkJoinPool().invoke(task);}
}

就像是递归二分法求和，不同的是这里左右分支使用线程去运行代码：

相对于串行需要注意的地方

线程安全问题，由于操作被分解由不同的线程执行，如使用共享的变量需要注意线程安全问题
执行顺序问题
代码说明

import java.util.Arrays;
import java.util.stream.Stream;public class Main {
    public static void main(String[] args) {
    System.out.println(Arrays.asList("a1", "a2", "b1", "c2", "c1").parallelStream().filter(s -> {
    System.out.format("filter: %s [%s]\n",s, Thread.currentThread().getName());return true;}).map(s -> {
    System.out.format("map: %s [%s]\n",s, Thread.currentThread().getName());return s.toUpperCase();}).reduce("",(a,b)->{
    System.out.format("reduce: %s %s [%s]\n",a,b, Thread.currentThread().getName());return  a+b;},(a,b)->{
    return  a+b;}));System.out.println(Stream.of(1,2,3,4,5).reduce(5,(a,b)->a+b,(a,b)->a+b));System.out.println(Stream.of(1,2,3,4,5).parallel().reduce(5,(a,b)->a+b,(a,b)->a+b));//分别是并行和非并行}
}

执行结果为
在这里插入图片描述
可以看出，代码的执行顺序已经被打乱，因此在执行严格顺序的业务时是不能使用并行流的。

上面代码有reduce的使用，需要注意的是，reduce第二个参数accumulator和第三个参数combiner的区别，accumulator是在并行流中串行计算结果的合并方法，combiner是并行进程计算结果的合并方法，这里需要注意，combiner应该与accumulator的计算方式相同。还有一个注意点，并行实际上就是分了好多个线程分别串行计算，注意：每个串行计算都会和idetity进行一次计算。因此需要保证identity对并行和串行计算的结果是相同的。例如求和identity应该为0否则在几个穿行计算中重复计算会导致数值不正确。