线程安全与否？从并行流更新非线程安全映射答案

【问题标题】：Thread safe or not? Updating a not-thread-safe-map from a parallel stream线程安全与否？从并行流更新非线程安全映射
【发布时间】：2016-03-16 10:34:41
【问题描述】：

下面的代码 sn-p 从并行流的 forEach 块更新非线程安全映射（itemsById 不是线程安全的）。

// Update stuff in `itemsById` by iterating over all stuff in newItemsById:
newItemsById.entrySet()
    .parallelStream()
    .unordered()
    .filter(...)
    .forEach(entry -> {
        itemsById.put(entry.getKey(), entry.getValue());      <-- look
    });

对我来说，这看起来不是线程安全的，因为并行流会同时在多个线程中调用forEach 块，从而在多个线程中同时调用itemsById.put(..)，并且@ 987654326@ 不是线程安全的。（但是，我认为使用 ConcurrentMap 代码是安全的）

我写信给同事：“请注意，当您插入新数据时，映射可能会分配新内存。这可能不是线程安全的，因为集合不是线程安全的。--无论是否写入不同的来自多个线程的键，是线程安全的，是依赖于实现的，我想。我不会选择依赖它。”

然而，他说上面的代码是线程安全的。 -- 是吗？

((请注意：我不认为这个问题过于本地化。实际上现在使用 Java 8，我认为相当多的人会做类似的事情：parallelStream()...foreach(...)，然后可能会很好地了解线程安全问题，因为很多人））

【问题讨论】：

如果有人说“这段代码是线程安全的”但没有解释为什么，你不应该相信他。即使某些东西本质上是安全的，例如因为它没有可变数据，这是你可以告诉别人的解释。关于任意Map 实现，您不需要在并发更新的情况下内存分配失败，对象结构的简单重新链接可能会导致任意错误。即使像增加 int 保持地图大小这样简单的操作也不是原子的，并且在并发执行时可能导致更新丢失，从而产生不一致的数据结构。

标签： java thread-safety java-stream

【解决方案1】：

您是对的：此代码不是线程安全的，并且取决于Map 实现和竞争条件可能会产生任何随机效果：正确的结果、无声的数据丢失、一些异常或无限循环。您可以像这样轻松地检查它：

int equal = 0;
for(int i=0; i<100; i++) {
    // create test input map like {0 -> 0, 1 -> 1, 2 -> 2, ...}
    Map<Integer, Integer> input = IntStream.range(0, 200).boxed()
         .collect(Collectors.toMap(x -> x, x -> x));
    Map<Integer, Integer> result = new HashMap<>();
    // write it into another HashMap in parallel way without key collisions
    input.entrySet().parallelStream().unordered()
         .forEach(entry -> result.put(entry.getKey(), entry.getValue()));
    if(result.equals(input)) equal++;
}
System.out.println(equal);

在我的机器上，这段代码通常会打印 20 到 40 而不是 100。如果我将 HashMap 更改为 TreeMap，它通常会因 NullPointerException 而失败，或者卡在 TreeMap 实现中的无限循环中。

【讨论】：

【解决方案2】：

我不是流方面的专家，但我认为这里没有采用花哨的同步，因此我不会考虑将元素并行添加到 itemsById 作为线程安全。

可能发生的事情之一是无限循环，因为如果两个元素碰巧最终在同一个存储桶中，则底层列表可能会混乱，并且元素可能会在一个循环中相互引用（A.next = B，B.next = A)。 ConcurrentHashMap 将通过同步存储桶上的写访问来防止这种情况发生，即除非元素最终位于同一存储桶中，否则它不会阻塞，但如果它们这样做，则添加是顺序的。

【讨论】：

【解决方案3】：

此代码不是线程安全的。

Oracle docs 状态：

像 forEach 和 peek 这样的操作是为副作用而设计的；一种返回 void 的 lambda 表达式，例如调用 System.out.println，除了副作用什么都做不了。即便如此，你应谨慎使用 forEach 和 peek 操作；如果你使用一个这些操作中的一个并行流，那么 Java 运行时可能调用您指定为其参数的 lambda 表达式 同时来自多个线程。

【讨论】：