并行流、收集器和线程安全

Question

请参阅下面的简单示例，该示例计算列表中每个单词的出现次数：

Stream<String> words = Stream.of("a", "b", "a", "c");
Map<String, Integer> wordsCount = words.collect(toMap(s -> s, s -> 1,
                                                      (i, j) -> i + j));

最后，wordsCount 是{a=2, b=1, c=1}。

但是我的流非常大，我想并行化这个工作，所以我写：

Map<String, Integer> wordsCount = words.parallel()
                                       .collect(toMap(s -> s, s -> 1,
                                                      (i, j) -> i + j));

但是我注意到wordsCount 是一个简单的HashMap，所以我想知道是否需要明确要求并发映射以确保线程安全：

Map<String, Integer> wordsCount = words.parallel()
                                       .collect(toConcurrentMap(s -> s, s -> 1,
                                                                (i, j) -> i + j));

非并发收集器可以安全地与并行流一起使用，还是应该在从并行流中收集时只使用并发版本？

Answer 1

非并发收集器可以安全地与并行流一起使用，还是应该在从并行流中收集时只使用并发版本？

在并行流的collect 操作中使用非并发收集器是安全的。

在Collector界面的specification，有六个要点的部分，是这样的：

对于非并发收集器，从结果提供者、累加器或组合器函数返回的任何结果都必须是串行线程限制的。这使得收集可以并行发生，而收集器不需要实现任何额外的同步。归约实现必须管理输入被正确分区，分区被隔离处理，并且只有在累加完成后才会合并。

这意味着Collectors 类提供的各种实现可以与并行流一起使用，即使其中一些实现可能不是并发收集器。这也适用于您可能实现的任何您自己的非并发收集器。它们可以安全地与并行流一起使用，前提是您的收集器不干扰流源、无副作用、与顺序无关等。

我还建议阅读 java.util.stream 包文档的Mutable Reduction 部分。本节中间是一个示例，该示例被声明为可并行化，但它将结果收集到 ArrayList 中，这不是线程安全的。

其工作方式是，以非并发收集器结束的并行流确保不同的线程始终在中间结果集合的不同实例上运行。这就是为什么收集器有一个Supplier 函数，用于创建与线程一样多的中间集合，因此每个线程都可以累积成自己的。当要合并中间结果时，它们会在线程之间安全地传递，并且在任何给定时间只有一个线程正在合并任何一对中间结果。

Answer 2

在并行流中使用非并发集合和非原子计数器是安全的。

如果您查看Stream::collect 的文档，您会发现以下段落：

与reduce(Object, BinaryOperator) 一样，收集操作可以并行化，无需额外同步。

对于Stream::reduce的方法：

虽然与简单地在循环中改变运行总数相比，这似乎是一种更迂回的方式来执行聚合，但归约操作可以更优雅地并行化，无需额外的同步，并且大大降低了数据争用的风险。

这可能有点令人惊讶。但是请注意，并行流是基于 fork-join 模型。这意味着并发执行如下：

• 将序列分成大小大致相同的两部分
• 单独处理每个部分
• 收集两个部分的结果并将它们合并为一个结果

在第二步中，三个步骤递归地应用于子序列。

一个例子应该可以清楚地说明这一点。

IntStream.range(0, 4)
    .parallel()
    .collect(Trace::new, Trace::accumulate, Trace::combine);

Trace 类的唯一目的是记录构造函数和方法调用。如果执行此语句，它会打印以下行：

thread:  9  /  operation: new
thread: 10  /  operation: new
thread: 10  /  operation: accumulate
thread:  1  /  operation: new
thread:  1  /  operation: accumulate
thread:  1  /  operation: combine
thread: 11  /  operation: new
thread: 11  /  operation: accumulate
thread:  9  /  operation: accumulate
thread:  9  /  operation: combine
thread:  9  /  operation: combine

可以看到，已经创建了四个 Trace 对象，accumulate 已经在每个对象上调用了一次，并且已经使用了 combine三次将四个对象合二为一。每个对象一次只能被一个线程访问。这使得代码是线程安全的，同样适用于 Collectors::toMap 方法。

Answer 3

所有收集器，如果它们遵循规范中的规则，则可以安全地并行或顺序运行。并行准备是这里设计的关键部分。

并发和非并发收集器之间的区别与并行化方法有关。

普通（非并发）收集器通过合并子结果进行操作。所以源被分割成一堆块，每个块被收集到一个结果容器中（如列表或地图），然后将子结果合并到一个更大的结果容器中。这是安全的并且可以保持顺序，但是对于某些类型的容器——尤其是地图——可能会很昂贵，因为通过键合并两个地图通常很昂贵。

并发收集器创建一个结果容器，它的插入操作保证是线程安全的，并从多个线程将元素爆炸到其中。对于像 ConcurrentHashMap 这样的高并发结果容器，这种方法可能比合并普通的 HashMap 表现得更好。

因此，并发收集器是对其普通对应物的严格优化。他们不是没有代价的。因为元素是从许多线程中被爆破的，并发收集器通常不能保持遇到顺序。 （但是，通常你并不关心——在创建字数直方图时，你并不关心你首先计算了哪个“foo”实例。）