java - 在每个元素被消耗后向 Java Stream 添加副作用？

Question

这是一个简化的示例，但它说明了要点。

假设我有一个如下定义的方法:

Stream<String> generateStream() {
  return Stream.of("hello", "world"); // (S)
}

有没有办法让S例如在元素被消耗之后打印一些东西，而流的使用者不必做或知道任何事情？

例如，我要修改 S 以便:

generateStream().forEach(System.out::println)

实际上将其打印到控制台:

hello
consumed
world
consumed

这在 Java 8 中可能吗？如果可以的话，如何实现？

Answer 1

您可以将 flatMap 与关闭操作一起使用:

Stream<String> generateStream() {
  return Stream.of("hello", "world")
      .flatMap(s -> Stream.of(s).onClose(() -> System.out.println(s+" consumed")));
}

按预期工作:

generateStream().forEach(System.out::println);

hello
hello consumed
world
world consumed

即使对于短路操作:

Optional<String> o = generateStream().filter(s -> s.startsWith("he")).findFirst();
o.ifPresent(s -> System.out.println("found "+s));

hello consumed
found hello

但请注意，与没有 flatMap 的操作相比，这会对性能产生巨大影响。但出于调试目的，或者对于性能不重要的情况，这可能会有所帮助。

此外，请记住，被 Stream 管道消耗并不意味着终端操作不会保留对它的引用或停止访问它。

对于forEach操作来说，它是这样工作的，但是例如，对于reduce((a,b) -> a)，所有元素都会被一个接一个地消耗掉，但对第一个元素的引用将保留到最后，甚至作为最终结果返回。对于min​(比较器)，任何元素都可能被保留，直到遇到更小的元素。最后，像 toArray() 这样的操作保存并返回结果中的所有元素。

此外，有状态操作可能会将后续管道处理与源流分离。例如。 sorted 步骤可能像 toArray 一样，在对数组进行排序并通过流式传输数组继续之前，缓冲所有元素，使它们显示为源已使用的元素。同样，distinct 将保留对超出其消耗的对象的引用，并且在并行流中，它们的后续处理可能会推迟到关闭源流后的某个点。