平面映射中引发的异常被onErrorResume运算符忽略

Question

考虑以下代码：

@Slf4j
@ExtendWith(MockitoExtension.class)
class ConnectionEventsConsumerTest {

    @Test
    public void testOnErrorResume() {
        Flux.range(0, 5)
                .doOnNext(event -> log.info("Processing -  {}", event))
                .flatMap(event -> processEvent(event)
                        .doOnSuccess(result -> log.info("Processed - {}", event))
                        .onErrorResume(t -> handleError(t, event))
                )
                .doOnError(t -> log.error("Exception propagated", t))
                //.log()
                .then()
                .subscribe();
    }

    private Mono<Void> processEvent(Object object) {
        return Mono.error(() -> new RuntimeException("test"));
        //throw new RuntimeException("test");
    }
    
    private Mono<Void> handleError(Throwable throwable, Object object) {
        log.error("Processing Failed - {}", object);
        
        return Mono.empty();
    }
    
}

如果方法Process Event返回Mono.error则输出与抛出Exception完全不同。

代码是这样的（返回一个Mono.error），我看到了我所期望的，300次处理和过程迭代失败，我看到没有异常传播。

17： 33:19.853[主]信息通信。playtika。服务。卡特福斯。pvp。服务Kafka。联系连接事件消费测试

• 处理-0 17:33:19.864[主]错误com。playtika。服务。卡特福斯。pvp。服务Kafka。联系连接事件消费测试
• 处理失败-0 17:33:19.865[主]信息com。playtika。服务。卡特福斯。pvp。服务Kafka。联系连接事件消费测试
• 处理-1 17:33:19.866[主]错误com。playtika。服务。卡特福斯。pvp。服务Kafka。联系连接事件消费测试
• 处理失败-1 17:33:19.866[主]信息com。playtika。服务。卡特福斯。pvp。服务Kafka。联系连接事件消费测试
• 处理-2 17:33:19.866[主]错误com。playtika。服务。卡特福斯。pvp。服务Kafka。联系连接事件消费测试
• 处理失败-2 17:33:19.866[主]信息com。playtika。服务。卡特福斯。pvp。服务Kafka。联系连接事件消费测试
• 处理-3 17:33:19.866[主]错误com。playtika。服务。卡特福斯。pvp。服务Kafka。联系连接事件消费测试
• 处理失败-3 17:33:19.866[主]信息com。playtika。服务。卡特福斯。pvp。服务Kafka。联系连接事件消费测试
• 处理-4 17:33:19.866[主]错误com。playtika。服务。卡特福斯。pvp。服务Kafka。联系连接事件消费测试
• 处理失败-4

另一方面，如果我取消对抛出的注释，我会看到正在处理的通量中的单个项，我看不到来自handleError的消息，我会看到“传播的异常”

17:35:53.950[main]信息com.playtika.services.catforce.pvp.service.kafka.connection.连接事件消费者测试

• 处理- 0 17:35:53.968[main]ERRORcom.playtika.services.catforce.pvp.service.kafka.connection.ConnectionEvents消费者测试
• 异常传播java.lang.RuntimeException： test

如果这是设计的，那么平面图的最佳实践是什么？想到的一个简单的解决方案是用try-catch将平面图的th内容包围起来，将异常封装在Mono中。错误虽然它可以工作，但它不雅观，太手工，很可能会被遗忘。

Answer 1

创建/返回Mono的方法不应该以这种方式抛出异常。由于异常是在组装（创建）Mono之前抛出的，因此平面图中的后续运算符不可能生效，因为它们需要Mono来操作。

如果您无法控制processEvent（）方法来修复其行为，那么可以用Mono将其包装起来。延迟，这将确保即使在装配期间出现的错误也会通过平面图内的单声道传播：

Flux.range(0, 5)
    .doOnNext(event -> log.info("Processing -  {}", event))
    .flatMap(event -> Mono.defer(() -> processEvent(event))
                .doOnSuccess(result -> log.info("Processed - {}", event))
                .onErrorResume(t -> handleError(t, event)))
    .doOnError(t -> log.error("Exception propagated", t))


private Mono<Void> processEvent(Object object) {
    throw new RuntimeException("test");
}

请注意，在其他中间操作符（如map或doOnNext）中，您可以随意以丑陋的方式抛出异常，因为Reactor可以将它们转换为适当的错误信号，因为此时单声道已经在进行中。