Project Reactor中如何将错误记录到MongoDB并保持错误传播链路正常

现有实现的问题 #

你当前的实现是不正确的：

• Mono.block()是阻塞操作，会直接打断响应式非阻塞的执行模型，占用线程资源导致吞吐量下降。如果当前执行线程是Netty EventLoop这类不允许阻塞的线程，甚至会直接抛出IllegalStateException异常。
• 阻塞操作也会拖慢整个响应式流水线的执行效率，不符合Reactor的设计规范。

是否可以改用Mono.subscribe()实现？ #

不推荐直接在doOnError中调用subscribe做错误落库：

• 单独subscribe的落库操作会脱离主流水线的生命周期控制，主流水线取消、终止时，落库操作仍可能在后台执行，造成资源泄露。
• 如果落库操作自身抛出异常，该异常不会被主流水线捕获，会触发Reactor全局未捕获异常处理逻辑，极端情况下会导致进程退出。
• 无法保证落库操作执行完成后再传播原错误，可能出现错误已经抛出到上层，但落库还没完成的时序问题。

正确实现方案 #

推荐使用onErrorResume操作符，将落库操作串联进主响应式流水线，既可以保证非阻塞执行，也能保留原有的错误传播逻辑，示例代码如下：

testMono.onErrorResume(throwable -> {
    var errorObject = ErrorObject.builder()
            .message(throwable.getLocalizedMessage())
            .className(throwable.getClass().getSimpleName())
            .build();
    // 将落库操作和原错误传播串联，保证落库完成后再抛出原错误
    return errorMessageRepository.save(errorObject)
            // 即使落库失败也不影响原错误抛出，避免覆盖业务原始错误
            .onErrorResume(saveErr -> Mono.empty())
            // 落库完成后返回携带原错误的Mono，保持原有错误传播逻辑
            .then(Mono.error(throwable));
})

这个实现的优势：

• 全程无阻塞，符合响应式开发规范
• 落库操作完全受控于主流水线的生命周期
• 落库操作的异常不会影响原有业务错误的传播，同时你也可以根据需要扩展落库异常的处理逻辑
• 能保证落库操作执行完成后再向上传播原错误，避免时序问题

内容的提问来源于stack exchange，提问作者n0noob