Java 并行流 : how to wait for threads for a parallel stream to finish?

Question

所以我有一个列表，我从中获取并行流来填充 map ，如下所示:

Map<Integer, TreeNode> map = new HashMap<>();
List<NodeData> list = some_filled_list;

//Putting data from the list into the map
list.parallelStream().forEach(d -> {
                TreeNode node = new TreeNode(d);
                map.put(node.getId(), node);
            });

//print out map
map.entrySet().stream().forEach(entry -> {
     System.out.println("Processing node with ID = " + entry.getValue().getId());
                });

此代码的问题在于，当“放置数据”过程仍在进行时(因为它是并行的)， map 正在打印出来，因此， map 尚未收到列表中的所有元素.当然，在我的真实代码中，不仅仅是打印出 map ；我使用 map 来利用 O(1) 查找时间。

我的问题是:

1. 如何让主线程等待，以便在打印出 map 之前完成“放置数据”？我试图将“放入数据”放入线程 t 中，并执行 t.start() 和 t.join()，但这没有帮助。

2. 也许在这种情况下我不应该使用并行流？列表很长，我只是想利用并行度来提高效率。

Answer 1

对于此 list.parallelStream().forEach，您违反了 Stream 文档中明确规定的 side-effects 属性。

此外，当您说此代码是在“放置数据”过程仍在进行时打印出 map (因为它是并行的)，这是不正确的，因为 forEach 是一个终端操作，它将等待完成，直到它可以处理下一行。你可能会看到这样，因为你正在收集到一个非线程安全的 HashMap 并且一些条目可能不在该映射中......考虑其他方式，如果将来自多个线程的多个条目放入 HashMap 中会发生什么情况？好吧，很多事情都可能会中断，比如丢失条目、不正确/不一致的 map 等。

当然，将其更改为 ConcurrentHashMap 是可行的，因为它是线程安全的，但是您仍然违反了副作用属性，尽管是以一种“安全”的方式。

正确的做法是直接将 collect 到 Map 而无需使用 forEach:

Map<Integer, TreeNode> map = list.parallelStream()
        .collect(Collectors.toMap(
                NodeData::getId,
                TreeNode::new
        ));

这样，即使是并行处理，一切都会好起来的。请注意，您需要很多(数万个元素)才能从并行处理中获得任何可衡量的性能提升。