记录通量直到单声道完成

Question

我有以下：

Flux<String> flux = ...
Mono<Void> mono = ...


Mono<Void> combined = operation(flux, mono);

它们代表并行发生的操作。

现在，我想将 flux 发出的所有元素打印到 sysout 直到 mono 完成， 在这里使用什么正确的运算符？

我试过了：

final Disposable subscribe = flux.subscribe(System.out::println);
mono.doOnSuccessOrError((o, e) -> subscribe.dispose());

但如果感觉笨拙，我觉得可能有更好的方法来做到这一点。有吗？

Answer 1

所以，Project Reactor 提供了一些方法来记录发出的结果：

• doOnNext(Consumer onNext)

• subscribe(Consumer consumer)

• log()

你可以看到它是如何工作的。

代码示例：

Flux<String> stringFlux = Flux.just("one", "two", "three")
            .doOnNext(s -> System.out.println("On next: " + s));
    Mono<String> stringMono = Mono.just("four");

    stringFlux = stringFlux.concatWith(stringMono)
            .map(s -> s + " hundred")
            .log();

    stringFlux.subscribe(s -> System.out.println("On next subscriber: " + s));

结果：

15:32:18.984 [main] INFO reactor.Flux.Map.1 - onSubscribe(FluxMap.MapSubscriber)
15:32:18.986 [main] INFO reactor.Flux.Map.1 - request(unbounded)
On next: one
15:32:19.005 [main] INFO reactor.Flux.Map.1 - onNext(one hundred)
On next subscriber: one hundred
On next: two
15:32:19.005 [main] INFO reactor.Flux.Map.1 - onNext(two hundred)
On next subscriber: two hundred
On next: three
15:32:19.005 [main] INFO reactor.Flux.Map.1 - onNext(three hundred)
On next subscriber: three hundred
15:32:19.006 [main] INFO reactor.Flux.Map.1 - onNext(four hundred)
On next subscriber: four hundred
15:32:19.007 [main] INFO reactor.Flux.Map.1 - onComplete()

第一条消息是从doOnNext 的stringFlux 写入的，之后是log 的组合发布者，最后一条是subscribe。

附：您还可以记录其他事件，例如 OnError、OnComplete 等。