在使用每个元素后向 Java Stream 添加副作用？答案

【问题标题】：Add a side-effect to Java Stream after each element is consumed?在使用每个元素后向 Java Stream 添加副作用？
【发布时间】：2020-08-01 09:29:26
【问题描述】：

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

假设我有一个这样定义的方法：

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

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

例如，我要修改S，这样：

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

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

hello
consumed
world
consumed

这在 Java 8 中是否可行？如果可以，如何实现？

【问题讨论】：

也许你正在寻找peek
我可以使用“peek”将字符串打印到控制台，但我想将字符串（“consumed”）“自动”打印，而无需消费者使用 peek。并且必须在值被消耗之后完成。有点像 aspectj 中的 @After 方面。
流或生产者无法知道某个部分何时被消费，因为没有这样的信号从消费者返回到生产者。您只能观看数据流动。还要考虑缓冲。生产者可能已经完成，而消费者在那之后很长一段时间都在处理缓冲的内容。当您将数据复制到 U 盘时，您可能从文件管理器的进度条中知道这种效果。
@Stefan 我认为你的评论会是一个很好的答案:)
意图似乎是为每种类型的终端操作添加Consumer，对吗？这将如何解决每个终端操作持续存在的现有关注点分离。

标签： java java-stream

【解决方案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(comparator)，可能会保留任何元素，直到遇到较小的元素。最后，像toArray() 这样的操作会保存并返回结果中的所有元素。

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

【讨论】：

【解决方案2】：

您可以在链中使用 peek 来使用元素并返回 Stream。 forEach 正在关闭流


    Stream.of("hello", "world")
        .peek(System.out::println)
        .forEach(word -> System.out.println("consumed"));

或者您可以在 foreach 中链接消费者以获得您的预期行为。 Stream 在 api 中没有明确的方法来满足您的要求，您可以使用 peek 而不是 foreach 链接方法以不完成流，或在 foreach 中链接消费者

    Consumer<String> stringConsumer = System.out::println;
    Consumer<String> endConsumed = string -> System.out.println("consumed");
    Stream.of("hello", "world")
            .forEach(stringConsumer.andThen(endConsumed));

【讨论】：