FluxSink.next() 抛出错误/异常，如果用户在处理器中抛出错误答案

【问题标题】：FluxSink.next() throw error/exception, if error thrown in processor by userFluxSink.next() 抛出错误/异常，如果用户在处理器中抛出错误
【发布时间】：2020-09-28 10:53:06
【问题描述】：

   Flux.<Integer>push(sink -> {
        try {
            for (int i = 0; i < 10 && !sink.isCancelled(); i++) {
                sink.next(i);
            }
            sink.complete();
        } catch (Exception e) {
            System.out.println("Error in wrong place " + e);
        }
    })
            .doOnNext(i -> { if (i % 5 == 0) throw new RuntimeException("Error"); })
            .subscribe(i -> System.out.println("i = " + i));

这段代码的输出是

  `Error in wrong place reactor.core.Exceptions$ErrorCallbackNotImplemented: java.lang.RuntimeException: Error`

表示sink.next(i);在 push 方法中重新抛出异常，而不是推送到错误反应通道。为什么？

【问题讨论】：

标签： reactive-programming project-reactor

【解决方案1】：

推送到flux错误频道，但是你在这个频道没有错误订阅者，你可以添加：

.doOnError(e -> System.out.println("Now error in good place = " + e))

或

.subscribe(i -> System.out.println("i = " + i), e -> System.out.println("Now error in good place = " + e));

那么完整的代码是：

   Flux.<Integer>push(sink -> {
        try {
            for (int i = 0; i < 10 && !sink.isCancelled(); i++) {
                sink.next(i);
            }
            sink.complete();
        } catch (Exception e) {
            System.out.println("Error in wrong place " + e);
        }
    })
            .doOnNext(i -> { if (i % 5 == 0) throw new RuntimeException("Error"); })
            .doOnError(e -> System.out.println("Now error in good place = " + e)) //it's missing
            .subscribe(i -> System.out.println("i = " + i));

还看到这个答案，为什么你需要订阅者和标准（狐狸 next()）subscriber 没有抛出错误：What is correct way to generate and handle exceptions when using Flux from projectreactor

【讨论】：