Stream<Stream>：flatMap 与 reduce [重复]

Question

如果我执行以下“连接”两个流的代码

• 首先通过平面映射Stream<Stream<Integer>>
• 然后通过使用Stream.concat() 减少Stream<Stream<Integer>>

两种情况我都得到了相同的正确结果，只是过滤操作的次数不同。

public class FlatMapVsReduce {
    public static void main(String[] args) {
        List<Integer> list = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9);

        Predicate<Integer> predicate1 = i -> {
            System.out.println("testing first condition with " + i);
            return i == 3;
        };

        Predicate<Integer> predicate2 = i -> {
            System.out.println("testing second condition with " + i);
            return i == 7;
        };

        System.out.println("Testing with flatMap");
        Integer result1 =
            Stream.of(list.stream().filter(predicate1),
                      list.stream().filter(predicate2))
                  .flatMap(Function.identity())
                  .peek(i -> System.out.println("peeking " + i))
                  .findFirst()
                  .orElse(null);
        System.out.println("result1 = " + result1);

        System.out.println();
        System.out.println("Testing with reduce");
        Integer result2 =
            Stream.of(list.stream().filter(predicate1),
                      list.stream().filter(predicate2))
                  .reduce(Stream::concat)
                  .orElseGet(Stream::empty)
                  .peek(i -> System.out.println("peeking " + i))
                  .findFirst()
                  .orElse(null);
        System.out.println("result2 = " + result2);
    }
}

我在两种情况下都得到了预期的结果 (3)。但是，第一个操作将第一个过滤器应用于集合的每个元素，而第二个操作在遇到一个时立即停止。输出是：

Testing with flatMap
testing first condition with 1
testing first condition with 2
testing first condition with 3
peeking 3
testing first condition with 4
testing first condition with 5
testing first condition with 6
testing first condition with 7
testing first condition with 8
testing first condition with 9
result1 = 3

Testing with reduce
testing first condition with 1
testing first condition with 2
testing first condition with 3
peeking 3
result2 = 3

为什么两者之间存在行为差异？ JDK 代码在第一个场景中是否可以改进为与在第二个场景中一样高效，或者 flatMap 中有什么东西使它不可能？

附录：以下替代方案与使用 reduce 的替代方案一样有效，但我仍然无法解释原因：

    Integer result3 = Stream.of(predicate1, predicate2)
                            .flatMap(c -> list.stream().filter(c).limit(1))
                            .peek(i -> System.out.println("peeking " + i))
                            .findFirst()
                            .orElse(null);
    System.out.println("result3 = " + result3);

Answer 1

从flatMap in openJDK的实现，我的理解是flatMap将传入流的全部内容推送到下游：

result.sequential().forEach(downstreamAsInt);

另一方面，Stream::concat 似乎正在处理拉取，而不是一次发送所有内容。

我怀疑你的测试没有显示完整的图片：

• 在flatMap 中，仅在第一个流耗尽时才考虑第二个流。
• 在reduce 中，所有流都被推送到最终连接的流中，因为在输入流的所有内容都被消耗之前，缩减的对象才有意义。

这意味着使用其中一个取决于您的输入有多复杂。如果你有一个无限的Stream<Stream<Integer>>，reduce 永远不会完成。