消耗每个元素后向 Flux 添加副作用？答案

【问题标题】：Add a side-effect to a Flux after each element is consumed?消耗每个元素后向 Flux 添加副作用？
【发布时间】：2020-11-22 00:04:27
【问题描述】：

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

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

Flux<String> generateFlux() {
  return Flux.just("hello", "world"); // (S)
}

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

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

generateFlux().doOnNext(System.out::println).block()

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

hello
consumed
world
consumed

是否可以使用 Reactor 3.3 做到这一点？如果可以，如何做到？

更新

我想我的问题不清楚，所以我会添加更多细节。我的问题的要点是我希望它不变：

generateFlux().doOnNext(System.out::println).block() // A

所以我想修改S，即Flux.just("hello", "world") 以适应我的要求，而不是A。我知道您可以将subscribe 和doOnNext 等添加到A，但这不是我要问的。我在问是否有办法修改S，以便在不对A 进行任何更改的情况下，当A 的订阅者/消费者处理完所有项目时会打印出来：

hello
consumed
world
consumed

即S 以某种方式接收到第一个元素已被消耗的信号（“hello”），然后它可以执行 doOnNext（或其他），在这种情况下打印“consumer”。当第二个元素被消耗（“world”）时，它应该再次打印“consumed”。

这可能吗？

【问题讨论】：

我会提供一些关于为什么这被否决的信息，因为我不知道我应该改变什么以使问题变得更好。

标签： java spring reactive-programming project-reactor

【解决方案1】：

Flux 是发布者，使用来自 Flux 的数据的正确方法是使用订阅。所以

 generateFlux.subscribe(word -> {
      System.out.println(word);
      System.out.println("consumed" + word);
});

使用 subscribe 并仅传递一个 Consumer 作为参数，您将覆盖订阅者接下来将执行的操作。基本上后面将创建一个订阅者，将这两个打印作为一种行为。 Flux 是如何工作的，你需要了解这 2 个接口

   public interface Publisher<T> {
       void subscribe(Subscriber<? super T> var1);
   }

   public interface Subscriber<T> {
       void onSubscribe(Subscription var1);
 
       void onNext(T var1);
 
       void onError(Throwable var1);

       void onComplete();
}

【讨论】：

【解决方案2】：

您可以使用 doOnNext 回调来拦截发布者推送下一个元素的动作。但使用订阅而不是阻止

generateFlux()
.doOnNext(next => System.out.println("Next Element pushed by publisher" + next))
.subscribe();

您可以使用覆盖 onNext、onError 和 onCompleted 的所有行为的订阅签名。 subscribe 方法在 Flux 中被重载

【讨论】：

【解决方案3】：

您可以使用doOnEach 为给定流中消耗的每个元素添加副作用，而无需修改发布者（错误和完成也是如此）。

  Flux.just("hello", "world")
                .doOnEach(item -> System.out.println(item))
                .blockLast();

doOnEach Documentation

还有很多针对不同生命周期的重载，例如 doOnError、doOnSuccess、doOnNext...

Documentation of Flux，你可以在那里找到所有的操作符和描述。

更新：

根据您的更新，您可能会想到横切关注点。您可以使用Hooks 来修改反应流操作符的组装时间。

Hooks Doc

【讨论】：