我如何知道 Java Stream collect(Collectors.toMap) 是否被并行化？答案

【问题标题】：How do I know if Java Stream collect(Collectors.toMap) is parallelized?我如何知道 Java Stream collect(Collectors.toMap) 是否被并行化？
【发布时间】：2015-12-05 00:06:33
【问题描述】：

我有以下代码尝试通过 Java Stream API 以并行方式从列表中填充地图：

class NameId {...}

public class TestStream
{
    static public void main(String[] args)
    {
        List<NameId > niList = new ArrayList<>();
        niList.add(new NameId ("Alice", "123456"));
        niList.add(new NameId ("Bob", "223456"));
        niList.add(new NameId ("Carl", "323456"));

        Stream<NameId> niStream = niList.parallelStream();
        Map<String, String> niMap = niStream.collect(Collectors.toMap(NameId::getName, NameId::getId));
    }
}

我如何知道地图是否使用多个线程（即并行）填充？我是否需要调用 Collectors.toConcurrentMap 而不是 Collectors.toMap？这是并行化地图人口的合理方法吗？我怎么知道支持新 niMap 的具体地图是什么（例如它是 HashMap）？

【问题讨论】：

n 是您的处理器拥有的核心数，将创建 n-1 个线程来对您的并行流进行操作。您的列表中有 3 行，因此您很可能会降低性能。
流框架故意向您隐藏实现（并行与否）。如果一切都做对了，就没有办法分辨了。

标签： java parallel-processing java-stream

【解决方案1】：

来自Javadoc：

返回的收集器不是并发的。对于并行流管道，组合器功能通过将键从一个映射合并到另一个映射来操作，这可能是一项昂贵的操作。如果不需要将结果按遇到顺序插入 Map，使用 toConcurrentMap(Function, Function) 可能会提供更好的并行性能。

所以听起来toConcurrentMap 将并行插入。

默认情况下，支持映射是HashMap。它只是调用toMap 的版本，它接受Supplier<M> 并传递HashMap::new。（来源：来源）

【讨论】：

【解决方案2】：

我如何知道地图是否使用多个线程（即并行）填充？

很难说。如果您的代码运行出奇地缓慢，可能是因为您尝试使用多个线程。

我是否需要调用 Collectors.toConcurrentMap 而不是 Collectors.toMap？

这将有助于提高并行效率，或者换一种说法，降低效率。

这是并行化地图人口的合理方法吗？

您可以按照您的建议进行操作，但是您应该注意，启动一个新线程的成本比您在此处所做的所有事情都要昂贵，因此即使添加一个线程也会大大降低速度。

我如何知道支持新 niMap 的具体映射是什么（例如它是 HashMap）？

文档说您无法确定。我上次检查 toMap 时使用的是 HashMap，groupingBy 使用的是 LinkedHashMap，但你不能假设它是任何特定的 Map。

【讨论】：

niMap.getClass 不会告诉你使用的是哪个地图吗？
@Jean-FrançoisSavard 是的，但它可能是 Java 更新之间的不同实现，或者理论上取决于你如何称呼它。也就是说，如果你没有传递一个 emptyMap() 对象，或者一个它可能是一个 singletonMap() 你甚至不能假设它是可变的。
groupingBy 何时使用 LinkedHashMap？我从来没有见过这个。
我自己最近才看到这个。

【解决方案3】：

您可以将toConcurrentMap 用于顺序流和toMap 用于并行流。区别是

toConcurrentMap() 并行流通常比顺序流更快
toMap() 对于顺序流通常比并行流更快

如果你不知道你的流从哪里来并且想在这两种情况下让它更快，你可以这样写：

Map<String, String> niMap = niStream.collect(
    niStream.isParallel() ? 
        Collectors.toConcurrentMap(NameId::getName, NameId::getId) :
        Collectors.toMap(NameId::getName, NameId::getId)
);

不同之处在于toConcurrentMap() 是一个CONCURRENT 收集器，这意味着使用了可以从不同线程同时填充的并发数据结构（在当前实现中为ConcurrentHashMap）。对于顺序流，这会增加一些不必要的开销，但对于并行流，它比使用 toMap() 更快，因为在 toMap() 情况下，将为每个并行线程创建单独的非并发 Map 实例，然后将这些 Map 合并在一起，这不是很好大地图的速度很快。

请注意，我的增强标准 Stream API 的 StreamEx 库添加了一个 toMap() 方法，该方法对并行流使用并发收集，对顺序流使用非并发收集：

Map<String, String> niMap = StreamEx.of(niStream)
                      .toMap(NameId::getName, NameId::getId);

【讨论】：