在3号Reactor抛出多个错误

Question

    Flux.create(sink -> {
        sink.error(new IOException());
        sink.error(new IOException());
    }).subscribe(System.out::println, System.out::println, System.out::println);

我（错误地）认为在第一个错误之后，通量将被处理，不再监听任何信号，但上面的代码创建了一个异常：

java.io.IOException
Exception in thread "main" reactor.core.Exceptions$BubblingException: java.io.IOException
    at reactor.core.Exceptions.bubble(Exceptions.java:154)
    at reactor.core.publisher.Operators.onErrorDropped(Operators.java:256)
    at reactor.core.publisher.FluxCreate$SerializedSink.error(FluxCreate.java:172)
    at main.App.lambda$f1$0(App.java:30)
    at reactor.core.publisher.FluxCreate.subscribe(FluxCreate.java:92)
    at reactor.core.publisher.Flux.subscribe(Flux.java:6447)
    at reactor.core.publisher.Flux.subscribeWith(Flux.java:6614)
    at reactor.core.publisher.Flux.subscribe(Flux.java:6440)
    at reactor.core.publisher.Flux.subscribe(Flux.java:6404)
    at main.App.f1(App.java:31)
    at main.App.main(App.java:23)
Caused by: java.io.IOException
    ... 8 more

为什么我有这个例外？

Answer 1

根据反应流，向onError（）发送两次信号是非法的，当create操作符收到第二个异常时，它会删除它。有一个钩子可以用来捕获这样的情况，并用它做一些有意义的事情，但是在您的情况下，默认的实现会起作用，这只是简单地包装无关的异常并将其抛出，或者“冒泡”给调用方。