关于Java8 parallelStream并发安全的深入讲解

Java 8引入的Stream API提供了一种非常方便和高效的处理集合的方式。parallelStream()方法可以使用多线程来利用CPU的多核执行计算。本文将深入讲解Java 8中parallelStream()的实现原理以及如何保证并发安全。

parallelStream() 并行流的实现原理

当调用parallelStream()方法时，底层机制会自动将单个流切分为多个小流，并将其分配给ForkJoinPool中的一组工作线程进行并行处理。ForkJoinPool是Java 7中新增的线程池框架，用途是完成”分而治之“（分治）任务处理。

在使用parallelStream()时，每个小流都被分配给一个独立的线程进行处理，这些线程在该分支中并行运行，并在结束时提供结果，然后合并结果以形成最终的输出。

由于parallelStream()底层使用ForkJoinPool来实现并行处理，因此其默认使用Runtime.getRuntime().availableProcessors()返回的CPU核心数量作为线程数。

在特定情况下，parallelStream()会降低性能。 例如，如果遇到进一步拆分数据的成本很高或数据太小而无法合理划分成小块时，它就可能被硬编码为单线程处理。

parallelStream()的三大要素

使用parallelStream()并行流具有三个要素：

• 拆分: 如果数据无法进一步拆分，或者拆分很昂贵，则该方法不会创建任何新线程。

• 执行: 如果上述步骤确定的是有效的，那么就在多个处理器上分发并行运行任务的一小部分。

• 结合: 最后一步是将处理器的结果组合在一起以形成答案。

如何保证并发安全？

在 parallelStream() 中我们可以通过以下方式来保证并发安全：

1. 使用不可变类

不可变类是指其对象创建后就不能在进行修改的类。在并行流处理时，不可变类是一种非常可靠的安全策略。多个线程可以同时读取该对象而不必担心对其造成影响。

例如一个字符串的操作：

List<String> words = Arrays.asList("this", "is", "a", "sample", "sentence");
long count = words.parallelStream().filter(word -> word.startsWith("s")).count();

此处的words为不可变的List，因此通过parallelStream()并行执行的过滤操作不会对words进行改变，保证了并发安全。

2. 使用同步方法

如果我们处理的是可变的对象，那么在处理之前我们可以通过同步方法保证对该对象的访问安全。每个线程都需要以线程安全的方式访问该对象，从而避免对该对象造成不一致的修改。

例如：

List<Integer> numbers = new ArrayList<>();
IntStream.range(0, 1000000).forEach(numbers::add);

synchronized (numbers) {
    int sum = numbers.parallelStream().mapToInt(num -> num).sum();
    System.out.println("Sum: " + sum);
}

在同步代码块中，我们可以避免并发访问number对象并保证计算结果的正确。

3. 使用线程安全的类

Java 8提供了线程安全的类，我们可以直接使用这些类避免对于线程安全的问题。例如，当我们需要对可变对象进行修改时，可以使用并发安全的数据结构类。

例如，当我们需要对可变的List进行修改时我们可以使用CopyOnWriteArrayList，它是一种线程安全的List实现，内部使用CopyOnWriteArray实现并发安全。

CopyOnWriteArrayList<Integer> numbers = new CopyOnWriteArrayList<>();

IntStream.range(0, 1000000).forEach(numbers::add);

int sum = numbers.parallelStream().mapToInt(num -> num).sum();
System.out.println("Sum: " + sum);

由于CopyOnWriteArrayList是一个线程安全的可变List类型对象，因此可以在parallelStream()中保证线程安全。

结论

Java 8中parallelStream()并行流为我们提供了非常方便和高效的处理集合的方式，但是在并行处理时必须小心以避免并发安全问题。使用不可变类、同步方法和线程安全的类可以帮助我们处理并发问题。