并行流与Fork/Join框架：如何使用并行流（Parallel Stream）提高大数据集合处理性能？

猫头虎

发布于 2024-12-18 18:55:30

12500

代码可运行

文章被收录于专栏：猫头虎博客专区猫头虎博客专区

运行总次数：0

代码可运行

并行流与Fork/Join框架：如何使用并行流（Parallel Stream）提高大数据集合处理性能？

引言

在大数据集合处理中，传统单线程操作可能导致性能瓶颈。JDK 8引入了并行流（Parallel Stream），基于Fork/Join框架，可以轻松实现数据的并行处理，大大提升执行效率。在本篇文章中，猫头虎将详细解析：

什么是并行流？如何与串行流（Stream）对比？
Fork/Join框架的底层原理是什么？
如何通过并行流提高大数据集合的处理性能？

学会并行流，让你的代码在处理大数据时飞起来！🚀

Java进阶之路：必知必会的核心知识点与JDK8、JDK17、JDK21版本对比

正文

问题背景：痛点描述

粉丝提问：猫哥，处理几百万条数据时，Stream效率很低怎么办？听说并行流可以提高性能，怎么用？
猫头虎解析：并行流通过将数据划分为多个任务，并利用多核CPU进行并发计算，大幅提高数据处理性能！

核心概念：什么是并行流？

1. 并行流与串行流的区别

类型	执行方式	适用场景
串行流	单线程逐个处理数据	适合小数据量、低并发场景。
并行流	将数据拆分成多个任务并行执行	适合大数据量、CPU密集型任务。

2. 并行流的实现原理

并行流底层基于Fork/Join框架，通过以下步骤实现并行处理：

任务拆分：将大数据分成多个小任务。
任务并行执行：将任务分发给多个线程并行计算。
任务合并：将并行结果合并，返回最终结果。

实战：如何使用并行流提高性能？

1. 并行流的基本使用

使用parallelStream()生成并行流，或者调用stream().parallel()将串行流转换为并行流。

示例代码 🚀

import java.util.Arrays;
import java.util.List;

public class ParallelStreamExample {
    public static void main(String[] args) {
        List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);

        // 使用串行流
        System.out.println("串行流结果:");
        numbers.stream()
               .filter(n -> n % 2 == 0)
               .forEach(n -> System.out.println(Thread.currentThread().getName() + " -> " + n));

        System.out.println("并行流结果:");
        // 使用并行流
        numbers.parallelStream()
               .filter(n -> n % 2 == 0)
               .forEach(n -> System.out.println(Thread.currentThread().getName() + " -> " + n));
    }
}

输出结果（部分）：

串行流结果:
main -> 2
main -> 4
main -> 6
main -> 8
main -> 10

并行流结果:
ForkJoinPool.commonPool-worker-3 -> 2
ForkJoinPool.commonPool-worker-5 -> 4
ForkJoinPool.commonPool-worker-1 -> 6
ForkJoinPool.commonPool-worker-7 -> 8
ForkJoinPool.commonPool-worker-3 -> 10

解读：

串行流使用单线程main执行任务。
并行流使用ForkJoinPool中的多个线程，实现并行处理。

2. Fork/Join框架的底层原理

Fork/Join框架是JDK 7引入的并行计算框架，分为以下步骤：

Fork：将任务拆分成多个子任务。
Join：将子任务执行的结果合并为最终结果。

核心类：

ForkJoinPool：线程池，用于执行任务。
RecursiveTask：用于有返回值的任务。
RecursiveAction：用于无返回值的任务。

示例代码 🚀

import java.util.concurrent.RecursiveTask;
import java.util.concurrent.ForkJoinPool;

public class ForkJoinExample extends RecursiveTask<Long> {
    private final long start;
    private final long end;
    private final long THRESHOLD = 10000;

    public ForkJoinExample(long start, long end) {
        this.start = start;
        this.end = end;
    }

    @Override
    protected Long compute() {
        if ((end - start) <= THRESHOLD) {
            long sum = 0;
            for (long i = start; i <= end; i++) {
                sum += i;
            }
            return sum;
        } else {
            long mid = (start + end) / 2;
            ForkJoinExample leftTask = new ForkJoinExample(start, mid);
            ForkJoinExample rightTask = new ForkJoinExample(mid + 1, end);

            leftTask.fork();  // 拆分任务
            rightTask.fork();
            return leftTask.join() + rightTask.join();  // 合并结果
        }
    }

    public static void main(String[] args) {
        ForkJoinPool pool = new ForkJoinPool();
        ForkJoinExample task = new ForkJoinExample(1, 1_000_000);
        long result = pool.invoke(task);
        System.out.println("计算结果: " + result);
    }
}

解读：

将计算1到100万的任务拆分成多个子任务并行计算。
Fork/Join框架通过Fork拆分任务，Join合并结果。

并行流的注意事项

线程开销：并行流适用于大数据量，小数据量场景下，线程切换会增加开销。
共享资源竞争：并行流可能导致数据竞争，需确保操作无副作用。
Fork/Join线程池：默认使用ForkJoinPool.commonPool，可以自定义线程池大小以提升性能。

Q&A互动答疑

Q：并行流适合哪些场景？ A：适合处理大数据量集合，并且计算过程是CPU密集型且无副作用的操作。

Q：如何避免并行流导致的线程安全问题？ A：避免在中间操作中修改共享数据，保证操作是无状态且无副作用的。

总结：掌握并行流的使用与底层原理

特性	优势	注意事项
并行流	提升大数据量处理性能，充分利用多核CPU。	小数据量不适用，需避免数据竞争。
Fork/Join框架	任务拆分与合并，高效实现并行计算。	自定义线程池需注意资源管理。