以下代码片段是获取目录列表、对每个文件调用提取方法并将生成的药物对象序列化为 xml 的方法的一部分。

try(Stream<Path> paths = Files.list(infoDir)) {
    paths
        .parallel()
        .map(this::extract)
        .forEachOrdered(drug -> {
            try {
                marshaller.write(drug);
            } catch (JAXBException ex) {
                ex.printStackTrace();
            }
        });
}

这是完全相同的代码，做完全相同的事情，但使用普通的 .list() 调用来获取目录列表并调用 .parallelStream()结果列表。

Arrays.asList(infoDir.toFile().list())
    .parallelStream()
    .map(f -> infoDir.resolve(f))
    .map(this::extract)
    .forEachOrdered(drug -> {
        try {
            marshaller.write(drug);
        } catch (JAXBException ex) {
            ex.printStackTrace();
    }
});

我的机器是四核 MacBook Pro，Java v 1.8.0_60(内部版本 1.8.0_60-b27)。

我正在处理 ~ 7000 个文件。 3 次运行的平均值:

第一个版本: 使用 .parallel():20 秒。没有 .parallel():41 秒

第二个版本: 使用 .parallelStream():12 秒。使用 .stream():41 秒。

鉴于 extract 方法从流中读取并完成所有繁重的工作，而 write 调用执行最终写入没有改变。

最佳答案

问题是 Stream API 的当前实现以及当前用于未知大小源的 IteratorSpliterator 实现严重地将此类源拆分为并行任务。您很幸运拥有超过 1024 个文件，否则您将完全没有并行化优势。当前的 Stream API 实现考虑了从 Spliterator 返回的 estimateSize() 值。未知大小的 IteratorSpliterator 在拆分前返回 Long.MAX_VALUE，其后缀也总是返回 Long.MAX_VALUE。其拆分策略如下:

1. 定义当前批量大小。当前的公式是从 1024 个元素开始，然后以算术方式增加(2048、3072、4096、5120 等)，直到达到 MAX_BATCH 大小(即 33554432 个元素)。
2. 将输入元素(在您的情况下为路径)消耗到数组中，直到达到批量大小或输入耗尽。
3. 返回一个 ArraySpliterator 迭代创建的数组作为前缀，将其自身作为后缀。

假设您有 7000 个文件。 Stream API 要求估计大小，IteratorSpliterator 返回 Long.MAX_VALUE。好的，Stream API 要求 IteratorSpliterator 拆分，它从底层 DirectoryStream 收集 1024 个元素到数组并拆分到 ArraySpliterator(估计大小1024) 及其自身(估计大小仍为 Long.MAX_VALUE)。由于 Long.MAX_VALUE 远大于 1024，Stream API 决定继续拆分较大的部分，甚至不尝试拆分较小的部分。所以整体的 split 树是这样的:

                     IteratorSpliterator (est. MAX_VALUE elements)
                           |                    |
ArraySpliterator (est. 1024 elements)   IteratorSpliterator (est. MAX_VALUE elements)
                                           |        |
                           /---------------/        |
                           |                        |
ArraySpliterator (est. 2048 elements)   IteratorSpliterator (est. MAX_VALUE elements)
                                           |        |
                           /---------------/        |
                           |                        |
ArraySpliterator (est. 3072 elements)   IteratorSpliterator (est. MAX_VALUE elements)
                                           |        |
                           /---------------/        |
                           |                        |
ArraySpliterator (est. 856 elements)    IteratorSpliterator (est. MAX_VALUE elements)
                                                    |
                                        (split returns null: refuses to split anymore)

所以之后你有五个并行任务要执行:实际包含 1024、2048、3072、856 和 0 个元素。请注意，即使最后一个 block 有 0 个元素，它仍然报告它估计有 Long.MAX_VALUE 个元素，因此 Stream API 也会将它发送到 ForkJoinPool。糟糕的是 Stream API 认为进一步拆分前四个任务是无用的，因为它们的估计大小要小得多。所以你得到的是输入的非常不均匀的分割，它最多使用四个 CPU 内核(即使你有更多)。如果您的每个元素处理对任何元素都花费大致相同的时间，那么整个过程将等待最大部分(3072 个元素)完成。所以你可能拥有的最大加速是 7000/3072=2.28x。因此，如果顺序处理需要 41 秒，那么并行流将花费大约 41/2.28 = 18 秒(接近您的实际数字)。

您的解决方案完全没问题。请注意，使用 Files.list().parallel() 您还将所有输入 Path 元素存储在内存中(在 ArraySpliterator 对象中) .因此，如果您手动将它们转储到 List 中，您将不会浪费更多内存。像 ArrayList(目前由 Collectors.toList() 创建)这样的支持数组的列表实现可以毫无问题地均匀拆分，从而进一步加快速度。

为什么这种情况没有优化？当然这不是不可能的问题(尽管实现起来可能非常棘手)。对于 JDK 开发人员来说，这似乎不是高优先级的问题。在邮件列表中有几个关于这个主题的讨论。您可以阅读 Paul Sandoz 消息 here他对我的优化工作发表评论。

关于java - 为什么 Files.list() 并行流的执行速度比使用 Collection.parallelStream() 慢得多？，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/34341656/