并行流

Question

有一个函数，可以在并行计算中计算出最常见的名字(Human[] people)。但是会存在数据竞争问题。为什么呢？

    Map<String, Integer> nameMap = new ConcurrentHashMap<>();
        Arrays.stream(people)
                .parallel()
                .filter(p -> p.isAdult())
                .map(Human::getName)
                .forEach(p -> nameMap.put(p, nameMap.containsKey(p) ? nameMap.get(p) + 1 : 1));
        return nameMap.entrySet().parallelStream().max((entry1, entry2) -> entry1.getValue() > entry2.getValue() ? 1 : -1).get().getKey();

- KateS

Answer 1

因为你进行的是get操作，然后是递增操作，再接着是put操作；在这之间，有可能已经有人将该条目放入nameMap中了。

在此处，你可以使用原子的ConcurrentHashMap#merge或更好地使用Collectors.toConcurrentMap。 编辑 你可能可以更加清晰地表达它：

  Arrays.stream(people)
        .parallel()
        .filter(Human::isAdult)
        .collect(Collectors.groupingBy(Human::getName, Collectors.counting()))
        .entrySet()
        .stream()
        .max(Comparator.comparing(Entry::getValue))
        .map(Entry::getKey)
        .get();

请注意，我相信你根本不需要使用parallel。

- Eugene

---

我认为最好使用groupingByConcurrent收集器，因为简单的groupBy会使用更重的合并操作。不过需要进行测试。 - M. Prokhorov

@M.Prokhorov，由于下游收集器是counting()，因此合并每个公共键的单个添加操作可能比在累加键时允许争用要便宜得多。然而，不使用并行处理执行操作比两种并行变体都更快，因为您需要一个非常大的输入数据集才能从并行处理中获得好处。 - Holger

@Eugene，你可以用Entry.comparingByValue()替换Comparator.comparing(Entry::getValue)。 - Holger