Java 中的嵌套并行流执行 - findAny() 随机失败

Question

对于 same 输入，以下代码在每 10-15 次尝试中抛出 IllegalArgumentException：

        AllDirectedPaths<Vertex, Edge> allDirectedPaths = new AllDirectedPaths<>(graph);
        List<GraphPath<Vertex, Edge>> paths = allDirectedPaths.getAllPaths(entry, exit, true, null);


        return paths.parallelStream().map(path -> path.getEdgeList().parallelStream()
                .map(edge -> {
                    Vertex source = edge.getSource();
                    Vertex target = edge.getTarget();

                    if (source.containsInstruction(method, instructionIndex)) {
                        return source;
                    } else if (target.containsInstruction(method, instructionIndex)) {
                        return target;
                    } else {
                        return null;
                    }
                }).filter(Objects::nonNull)).findAny().flatMap(Stream::findAny)
                .orElseThrow(() -> new IllegalArgumentException("Given trace refers to no vertex in graph!"));

代码的想法是找到一个包装特定指令的顶点（参见containsInstruction()），而该顶点至少位于从entry 到exit 顶点的一条路径上。我知道代码在性能方面并不是最优的（路径上的每个中间顶点都被查找两次），但这并不重要。

输入只是一个跟踪（字符串），可以从中导出method 和instructionIndex。从这个意义上说，所有其他变量都是固定的。而且containsInstruction()这个方法没有任何副作用。

将“findAny()”流操作放在哪里重要吗？我应该把它直接放在过滤器操作之后吗？还是嵌套的并行流是问题所在？

Answer 1

您应该使用.flatMap(path -> ... ) 并删除.flatMap(Stream::findAny)。

您的代码不起作用，因为第一个 findAny() 返回一个始终为非 null 的流，但它可能包含 null 元素。

然后，当您通过Optional.flatMap(Stream::findAny) 调用应用第二个findAny() 时，最后一个查找操作可能会返回一个空的Optional，结果是内部流的null 元素结束.

这就是代码的样子：

return paths.stream()
    .flatMap(path -> path.getEdgeList().stream()
        .map(edge -> 
             edge.getSource().containsInstruction(method, instructionIndex) ?
             edge.getSource()                                               :
             edge.getTarget().containsInstruction(method, instructionIndex) ?
             edge.getTarget()                                               :
             null)
        .filter(Objects::nonNull))
    .findAny()
    .orElseThrow(() -> new IllegalArgumentException("whatever"));

请注意：为什么是并行流？您的管道中似乎没有 CPU 绑定任务。此外，并行流会产生大量开销。它们在极少数情况下很有用，即流水线中的数万个元素和密集的 CPU 操作

---

编辑： 正如 cmets 中所建议的，内部流的 map 和 filter 操作可以安全地移至外部流。这样，可读性得到了提高，并且在性能方面没有差异：

return paths.stream()
    .flatMap(path -> path.getEdgeList().stream())
    .map(edge -> 
         edge.getSource().containsInstruction(method, instructionIndex) ?
         edge.getSource()                                               :
         edge.getTarget().containsInstruction(method, instructionIndex) ?
         edge.getTarget()                                               :
         null)
    .filter(Objects::nonNull)
    .findAny()
    .orElseThrow(() -> new IllegalArgumentException("whatever"));

另一个注意事项：也许将map 中的代码重构为Edge 类的方法会更好，以便返回源、目标或null 的逻辑在已经有的类中所有信息。