FluxSink。next（）如果用户在处理器中引发错误，则引发错误/异常

Question

   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` 

这意味着异常将由<代码>接收器重新引发。下一步（i） 在推送方法和非推送错误反应通道中。为什么？

Answer 1

它被推送到流量错误通道，但您在此通道上没有错误订阅者，您可以添加：

.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));

另请参阅此答案，为什么您需要订阅者和标准（fox Next（））订阅者没有抛出错误：当使用来自项目处理者的Flux时，生成和处理异常的正确方法是什么