在并行流中的 hashmap 中插入值时的线程安全

Question

我需要进行超时为 10 秒的异步调用，并且需要对地图中的每个元素执行此操作。异步调用的结果存储在另一个映射中。在这种情况下使用 HashMap 是否安全，还是我需要使用 ConcurrentMap？

Map<String, String> x = ArrayListMultimap.create();
Map<String, Boolean> value = Maps.newHashMap();

x.keySet().paralleStream().forEach(req -> {
   try {
      Response response = getResponseForRequest(req);
      value.put(req, response.getTitle());
   } catch(TimeoutException e) {
      value.put(req, null);
   }
}

这个线程安全吗？我想不通。我知道另一种方法是创建一个并发哈希图，并考虑一些其他填充值而不是空值，因为并发映射不支持空值。

Answer 1

您可以使用 .map() 代替 .forEach() 和 return 使用 Collectors.toMap() 终止函数创建的地图，而不是并行修改外部地图。考虑以下示例：

Map result = x.keySet()
  .parallelStream()
  .map(req -> {
    try {
      Response response = getResponseForRequest(req);
      return new AbstractMap.SimpleEntry<>(req, response.getTitle());
    } catch (TimeoutException e) {
      return new AbstractMap.SimpleEntry<>(req, null);
    }
  })
  .collect(Collectors.toMap(AbstractMap.SimpleEntry::getKey, AbstractMap.SimpleEntry::getValue));

在此示例中，您 return 一个 SimpleEntry 对象，它表示每个元素的键和值，当处理完所有条目后，您将它们收集到一个映射中。

简化

Holger 建议完全去掉 AbstractMap.SimpleEntry 以简化解决方案：

Map result = x.keySet()
  .parallelStream()
  .collect(Collectors.toMap(Function.identity(), req -> {
    try {
      Response response = getResponseForRequest(req);
      return response.getTitle()
    } catch (TimeoutException e) {
      return null
    }
  }));

选择最适合你的。

在并行流中的 hashmap 中插入值时的线程安全

Thread safety while inserting values in hashmap in parallel stream

java

concurrency

hashmap

concurrenthashmap

java-8

简化