【Java8实战】并行流

约定不等于承诺〃 2023-07-01 12:57 95阅读 0赞

#### 引言 ####

在前面，学习使用Lambda表达式的过程中，对于集合的处理，都会使用到Stream流处理。为了提高性能，我们可以使用parallelStream并行流。

#### 并行流 ####

并行流就是一个把内容分成多个数据块，并用不同的线程分别处理每个数据块的流。

这样一来，你就可以自动把给定操作的工作负荷分配给多核处理器的所有内核，让它们都忙起来。

#### 实例代码 ####

比如，我们求1到n一组数的和，用传统的java代码实现，如下：

/**
     * 传统方式实现，1-10的求和
     * @param n
     * @return
     */
    public static long iterativeSum(long n) {    
     long result = 0;     
     for (long i = 1L; i <= n; i++) {         
     	result += i;    
      }     
      return result;
    }

而使用java8中新特性流的处理，可以写成如下代码：

/**
     * java8 Stream.iterate顺序流处理
     * @param n
     * @return
     */
     public static long sequentialSum(long n) {
            return Stream.iterate(1L, i -> i + 1).limit(n).reduce(0L, Long::sum);  
     }

当n特别大时，我们可以借助并行流处理，代码如下：

/**
     * java8 Stream.iterate并行流处理
     * @param n
     * @return
     */
    public static long parallelSum(long n) {
        return Stream.iterate(1L, i -> i + 1).limit(n).parallel().reduce(0L, Long::sum);
    }

#### 操作原理 ####

并行流的不同之处在于Stream在内部分成了几块，因此可以对不同的块独立并行进行归纳操作，最后，同一个归纳操作会将各个子流的部分归纳结果合并起来，得到整个原始流的归纳结果。

如下图所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMwMzQyMjM_size_16_color_FFFFFF_t_70]

### 线程池配置 ###

并行流用的线程是从哪儿来的？有多少个？怎么自定义这个过程呢？ 下面给出答案：

并行流内部使用了默认的ForkJoinPool，它默认的线程数量就是你的处理器数量，这个值是由Runtime.getRuntime().available- Processors()得到的。

可以通过系统属性java.util.concurrent.ForkJoinPool.common. parallelism来改变线程池大小，如下所示： System.setProperty(“java.util.concurrent.ForkJoinPool.common.parallelism”,“12”);

这是一个全局设置，因此它将影响代码中所有的并行流。一般而言，不建议对其进行修改。

#### 性能测试 ####

我们用下面一段代码，来输出10次内，求和执行时间最短的一次：

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;
    }

(1) **Sequential sum 顺序流求和耗时结果**：  
![在这里插入图片描述][20200122143205459.png]  
(2) **Iterative sum 传统迭代求和耗时结果：**  
![在这里插入图片描述][2020012214330683.png]  
用传统for循环的迭代版本执行起来应该会快很多，因为它更为底层，更重要的是不需要对 原始类型做任何装箱或拆箱操作。

(3) **Parallel sum 并行流求和耗时结果：**

![在这里插入图片描述][20200122143506177.png]  
看到结果有点让人失望了，求和方法的并行版本比顺序版本要慢很多。原因是：

*  iterate生成的是装箱的对象，必须拆箱成数字才能求和；
 *  我们很难把iterate分成多个独立块来并行执行。

这就说明了并行编程可能很复杂，有时候甚至有点违反直觉。如果用得不对（比如采用了一 个不易并行化的操作，如iterate），它甚至可能让程序的整体性能更差。

那么如何才能高效地使用并行流呢？流处理中为我们提供了LongStream.rangeClosed的方法。这个方法与iterate相比有两个优点：

*  LongStream.rangeClosed直接产生原始类型的long数字，没有装箱拆箱的开销。
 *  LongStream.rangeClosed会生成数字范围，很容易拆分为独立的小块。例如，范围1~20 可分为1-5、6-10、11-15和16-20。

下面看使用它用顺序流操作的结果：

/**
     * java8 LongStream.rangeClosed顺序流处理
     * @param n
     * @return
     */
    public static long rangedSum(long n) {
       return  LongStream.rangeClosed(1, n).reduce(0L, Long::sum);
    }

![在这里插入图片描述][20200122145410746.png]  
这个数值流比前面那个用iterate工厂方法生成数字的顺序执行版本要快得多，因为数值流避免了非针对性流那些没必要的自动装箱和拆箱操作。由此可见，选择适当的数据结构往往比并行化算法更重要。

如果使用并行流，结果会是怎样呢？

/**
     * java8 LongStream.rangeClosed并行流处理
     * @param n
     * @return
     */
    public static long parallelRangedSum(long n) {
        return LongStream.rangeClosed(1, n).parallel().reduce(0L, Long::sum);
    }

![在这里插入图片描述][20200122150139628.png]  
这一次结果，就和预期的一致了，这样的并行操作比顺序执行更快。这也表明，使用正确的数据结构然后使其并行工作能够保证最佳的性能。

#### 高效使用并行流 ####

对于并行流操作的使用，我们应该根据情况具体分析是否适合使用，并不是使用并行流就一定能够提高性能，使用不当，反而会有不好的结果。所以，下面的问题就是我们在使用并行流应注意哪些方面，才能达到高效。

(1) **测试**。把顺序流转成并行流轻而易举，但却不一定是好事。**我们应该基于合理的测量，根据结果决定来判断使用并行流是否合适。**

(2) **留意装箱**。自动装箱和拆箱操作会大大降低性能。Java 8中有原始类型流（IntStream、 LongStream、DoubleStream）来避免这种操作，但凡有可能都应该用这些流。

(3) **有些操作本身在并行流上的性能就比顺序流差**。特别是limit和findFirst等依赖于元 素顺序的操作，它们在并行流上执行的代价非常大。例如，findAny会比findFirst性 能好，因为它不一定要按顺序来执行。

(4) **对于较小的数据量，选择并行流几乎从来都不是一个好的决定**。并行处理少数几个元素 的好处还抵不上并行化造成的额外开销。

(5) **要考虑流背后的数据结构是否易于分解**。例如，ArrayList的拆分效率比LinkedList高得多，因为前者用不着遍历就可以平均拆分，而后者则必须遍历。

(6) **考虑终端操作中合并步骤的代价是大是小**（例如Collector中的combiner方法）。 如果这一步代价很大，那么组合每个子流产生的部分结果所付出的代价就可能会超出通过并行流得到的性能提升。

实例代码已更新至github，地址：[parallel stream并行流][parallel stream]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMwMzQyMjM_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200122133143324.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMwMzQyMjM=,size_16,color_FFFFFF,t_70
[20200122143205459.png]: https://img-blog.csdnimg.cn/20200122143205459.png
[2020012214330683.png]: https://img-blog.csdnimg.cn/2020012214330683.png
[20200122143506177.png]: https://img-blog.csdnimg.cn/20200122143506177.png
[20200122145410746.png]: https://img-blog.csdnimg.cn/20200122145410746.png
[20200122150139628.png]: https://img-blog.csdnimg.cn/20200122150139628.png
[parallel stream]: https://github.com/huzhiting/java8-learning.git