并行流、收集器和线程安全
请参阅下面的简单示例，该示例计算列表中每个单词的出现次数：
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));可以与并行流一起安全地使用非并发收集器，还是应该仅在从并行流进行收集时使用并发版本？
回答 3
Stack Overflow用户
发布于 2014-03-12 20:17:24
非并发收集器可以安全地与并行流一起使用吗?还是应该仅在从并行流收集时使用并发版本？
在并行流的collect操作中使用非并发收集器是安全的.
在规格说明的Collector接口中，在有六个要点的部分中，如下所示：
对于非并发收集器，从结果提供者、累加器或组合器函数返回的任何结果都必须是串行线程限制的。这使得集合可以并行进行，而无需实现任何额外的同步。还原实现必须管理输入被正确地分区，分区被隔离地处理，并且只有在积累完成后才会进行组合。
这意味着Collectors类提供的各种实现可以与并行流一起使用，即使其中一些实现可能不是并发收集器。这也适用于您自己可能实现的任何非并发收集器。它们可以安全地与并行流一起使用，只要您的收集器不干扰流源，没有副作用，命令无关，等等。
我还建议阅读java.util.stream包文档的java.util.stream部分。在本节的中间有一个例子，说明它是可并行的，但是它将结果收集到一个ArrayList中，这并不是线程安全的。
其工作方式是，以非并发收集器结尾的并行流确保不同的线程总是在中间结果集合的不同实例上操作。这就是为什么收集器有一个Supplier函数，用于创建尽可能多的中间集合，这样每个线程就可以累积到自己的线程中。当中间结果被合并时，它们在线程之间被安全地传递，并且在任何给定的时间，只有一个线程正在合并任何一对中间结果。
Stack Overflow用户
发布于 2014-04-29 16:36:32
如果所有收集器遵循规范中的规则，则所有收集器都可以安全地并行或顺序运行。并行准备是这里设计的关键部分。
并发收集器和非并发收集器之间的区别与并行化的方法有关。
普通(非并发)收集器通过合并子结果来操作。因此，源被划分成一组块，每个块被收集到一个结果容器中(比如一个列表或映射)，然后子结果被合并到一个更大的结果容器中。这是安全和维持秩序的，但对于某些类型的容器--特别是地图--可能很昂贵，因为按键合并两个映射通常很昂贵。
相反，并发收集器创建一个结果容器，其插入操作保证线程安全，并从多个线程向其发送元素。对于像ConcurrentHashMap这样高度并发的结果容器，这种方法可能比合并普通HashMaps更好。
因此，并发收集器比普通收集器严格优化。而且它们并不是没有代价的；因为元素是从许多线程中释放出来的，并发收集器通常无法保存遭遇顺序。(但是，通常您并不关心-在创建单词计数直方图时，您并不关心首先计算的是哪个"foo“实例。)
Stack Overflow用户
发布于 2014-03-12 20:22:18
使用具有并行流的非并发集合和非原子计数器是安全的.
如果您查看溪流：：收集的文档，您会发现以下段落：
与
reduce(Object, BinaryOperator)一样，可以并行化收集操作，而不需要额外的同步。
对于方法流：：减少
虽然这似乎是执行聚合的一种更迂回的方式，而不是简单地在循环中变异运行的总数，但是简化操作更优雅地并行化，而不需要额外的同步，并且大大降低了数据竞争的风险。
这可能有点令人吃惊。但是，请注意，并行流基于叉-连接模型。这意味着并发执行的工作如下：
- 将序列分成两部分，大小大致相同
- 单独处理每一部分
- 收集两个部分的结果，并将它们合并为一个结果。
在第二步中，这三个步骤被递归地应用于子序列。
一个例子应该说明这一点。这个
IntStream.range(0, 4)
.parallel()
.collect(Trace::new, Trace::accumulate, Trace::combine);类跟踪的唯一目的是记录构造函数和方法调用。如果执行此语句，它将打印以下行：
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您可以看到，已经创建了四个跟踪对象，在每个对象上调用了一次累积，并使用了三次组合将这四个对象组合为一个。每个对象一次只能由一个线程访问。这使得代码线程安全，这同样适用于收集器：：toMap方法。
https://stackoverflow.com/questions/22350288
复制相似问题
腾讯云开发者
