反应流中不一致的doAfterTerminate行为

Question

添加行为（副作用）在流量终止后触发，要么成功完成下游，要么出现错误。

这里有一个简单的例子来再现我面临的问题：

public static void main(String[] args) {
    boolean shouldThrow = true;
    Flux.just("string1" /* , "string2" */)
            // 1
            .materialize()
            // 2
            .doOnNext(s -> {
                if (shouldThrow)
                    throw new RuntimeException();
            })
            // 3
            .dematerialize()
            // 4
            .onErrorResume(t -> Flux.empty())
            .subscribe();
}

ShouldThrow标志简单地说明了在Materialize和Dematerialize调用之间的代码可能引发错误。数字1到4表示为了本例的目的，我尝试插入doAfterTerminate调用的位置，doAfterTerminate(()->system.out.println（“after Terminate”）。

null

有人能解释一下上面的事情吗？这是图书馆的故意行为吗？

注意，我使用的是Reactor堆芯3.3.8。

Answer 1

不是不一致的行为。

如你所知，通量发射0...n个元素。错误/完成信号关闭流。

当您实现时，真正发生的是，不是发送元素，而是发送包装在信号对象中的元素。流量 变为流量 > 。这甚至是一个完整的信号包装和发送作为一个对象在管道中，如图所示。

    boolean shouldThrow = false;
    Flux.just("string1" /* , "string2" */)
           //1
            .doOnNext(s -> System.out.println(s))
            .materialize()
            //2
            .handle(((stringSignal, synchronousSink) -> {
                synchronousSink.complete();
            }))
            .doOnNext(s -> {
                if (shouldThrow)
                    throw new RuntimeException();
            })
            .doAfterTerminate(() -> System.out.println("after terminate"))
            .dematerialize()
            // 4
            .onErrorResume(t -> Flux.empty())
            .subscribe();