并行数据处理与性能详解与ForkJoin框架

文章目录

一、并行流
- 1、将顺序流转换成并行流
- 2、测量流的性能
二、分之/合并框架ForkJoinPool
- 1、使用RecursiveTask
三、Spliterator

本章节可以让你用Stream接口不费力气就能对数据集执行并行操作，可以声明性的讲顺序流变成并行流。

一、并行流

Stream接口可以调用方法parallelStream很容易把集合转换为并行流。所谓并行流就是把内容分成多个数据块，用不同线程处理每块数据。

1、将顺序流转换成并行流

可以将流转换成并行流，调用方法parallel。例：

public static long parallelSum(long n) {return Stream.iterate(1L,i -> i+1).limit(n).parallel().reduce(0L,Long::sum);}

如果从并行流变成顺序流可以调用sequential这个方法完成。

stream.parallel()
.filter(...)
.sequential()
.map(...)
.parallel()
.reduce();

2、测量流的性能

按常理并行求和方法应该比迭代方法性能好。然而在软件工程上，靠猜是绝对不行的，因此我们要进行实战看结果。

public class ParallelStreams {public static long measureSumPerf(Function<Long, Long> adder, long n) {long fastest = Long.MAX_VALUE;for (int i = 0; i < 10; i++) {long start = System.nanoTime();long sum = adder.apply(n);long duration = (System.nanoTime() - start) / 1_000_000;System.out.println("Result: " + sum);if (duration < fastest) fastest = duration;}return fastest;}public static long iterativeSum(long n) {long result = 0;for (long i = 1L; i <= n; i++) {result += i;}return result;}public static long parallelSum(long n) {return Stream.iterate(1L, i -> i + 1).limit(n).parallel().reduce(0L, Long::sum);}public static long sequentialSum(long n) {return Stream.iterate(1L, i -> i + 1).limit(n).reduce(0L, Long::sum);}public static void main(String[] args) {System.out.println("Sequential sum done in:" +measureSumPerf(ParallelStreams::sequentialSum, 10_000_000) + " msecs");System.out.println("Iterative sum done in:" +measureSumPerf(ParallelStreams::iterativeSum, 10_000_000) + " msecs");System.out.println("Parallel sum done in: " +measureSumPerf(ParallelStreams::parallelSum, 10_000_000) + " msecs" );}
}

运行结果：

运行结果相当令人失望，求和方法并行是顺序版本的将近10倍，为什么会出现这样的结果，实际上有两个问题：
1、iterate生成是装箱的对象，必须拆箱才能求和。
2、很难把iterate分成多个独立块来并行执行。iterate很难分割成能独立执行的小块，因为每次应用这个函数都要依赖前一次应用执行结果。

使用有针对性的方法，避免装箱拆箱操作和能分成独立块并行。可以使用，LongStream.rangeClosed与iterate相比有两个优点。
1、产生原始long数字，没有装箱拆箱操作。
2、会生成数字范围，很容易拆成小块。
例：

 public static long rangedSum(long n) {return LongStream.rangeClosed(1, n).reduce(0L, Long::sum);}public static long parallelRangedSum(long n) {return LongStream.rangeClosed(1, n).parallel().reduce(0L, Long::sum);}System.out.println("Parallel sum done in: " +measureSumPerf(ParallelStreams::rangedSum, 10_000_000) + " msecs" );System.out.println("Parallel sum done in: " +measureSumPerf(ParallelStreams::parallelRangedSum, 10_000_000) + " msecs" );

执行结果：

得到结果终于比顺序执行快，我们使用并行流的时候一定要正确使用，比如算法改变了某些共享状态。

二、分之/合并框架ForkJoinPool

分之/合并框架的目的是有递归方式将可以并行的任务拆分成更小的任务，然后将每个子任务的结果合并起来生成整体结果。ForkJoinPool是ExecutorService的一个实现，它把子任务分配给线程池(称为ForkJoinPool)中的工作线程。

1、使用RecursiveTask

要把任务提交到线程池中，必须创建RecursiveTask的子类，重写compute方法。R是并行化产生的结果类型。如果任务不返回结果用RecursiveAction。
protected abstract R compute();这个方法同时定义了将任务拆分成子任务的逻辑，和任务无法在拆分时，生成单个任务逻辑。

if (任务足够小或不可分) {
顺序计算该任务
} else {
将任务分成两个子任务
递归调用本方法，拆分每个子任务，等待所有子任务完成
合并每个子任务的结果
}

用分之/合并框架并行求和

public class ForkJoinSumCalculator extends RecursiveTask<Long> {private final long[] numbers;private final int start;private final int end;public static final long THRESHOLD = 10_000;public ForkJoinSumCalculator(long[] numbers) {this(numbers,0,numbers.length);}private 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 leftResult + rightResult;}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);}public static void main(String[] args) {System.out.println(forkJoinSum(10000000));}
}

使用分之/合并框架需注意几点：
1、对于一个任务调用join方法会阻塞调用方，直到该任务结束。因此，在两个子任务的计算开始之后再调用它。
2、不在RecursiveTask子类内部使用invoke方法

三、Spliterator

Spliterator是java8中加入的另一个新接口，字面意思是可分迭代器，主要作用于并行执行。

public interface Spliterator<T> {、//按顺序一个一个使用Spliterator元素，如果有其它元素遍历返回trueboolean tryAdvance(Consumer<? super T> action);//可以把元素划分出去分给第二个SpliteratorSpliterator<T> trySplit();//还剩多少元素要遍历long estimateSize();//特性int characteristics();
}