Reactor中doOnError阻止异常传播，如何保留doOnError同时让异常被try-catch捕获

原理说明 #

1. Reactor 中 Mono/Flux 的异常属于流内的错误信号，和常规 JVM 栈抛出的异常逻辑不同，默认不会直接泄漏到订阅代码的外层栈空间。
2. 你移除 doOnError 之后异常能被外层 try-catch 捕获，本质是你调用的 subscribe 重载只声明了正常数据的消费逻辑，没有声明错误处理逻辑，旧版本 Reactor 会将未处理的异常包装后抛出到当前订阅线程，刚好被外层捕获。
3. 你代码里 FLOW FINISHED 没有打印是因为流发出的是错误信号，正常数据的消费逻辑不会被触发，只有流正常发出元素时才会执行该打印语句。
4. doOnError 是副作用操作符，只会监听错误信号，不会消费错误，错误信号会继续向下传递，不会导致异常消失。

解决方案 #

要同时保留 doOnError 且让外层 try-catch 捕获异常，可以使用以下两种方案：

方案1：使用 block() 阻塞等待流执行完成 #

block() 会同步等待流结束，流产生的异常会直接在当前线程抛出，可直接被外层 try-catch 捕获：

fun method(): Mono<String> {    
    return Mono.error(RuntimeException("RUNTIME EXCEPTION"))
}
fun main(args: Array<String>) {    
    try {        
        method()            
            .doOnError {                
                when(it){                    
                    is RuntimeException -> println("DO_ON_NEXT RUNTIME EXCEPTION")                    
                    else -> println("DO_ON_NEXT OTHER EXCEPTION")                
                }            
            }
            // 阻塞等待流执行完成，异常直接抛出
            .block()
        println("FLOW FINISHED")
    } catch (e: Exception){        
        println("CAUGHT EXCEPTION IN TRY CATCH CLAUSE")    
    }
}

方案2：订阅时显式传入错误回调，主动抛出异常 #

如果你需要使用 subscribe 而非阻塞调用，可以显式传入错误消费逻辑，在回调内重新抛出异常：

fun method(): Mono<String> {    
    return Mono.error(RuntimeException("RUNTIME EXCEPTION"))
}
fun main(args: Array<String>) {    
    try {        
        method()            
            .doOnError {                
                when(it){                    
                    is RuntimeException -> println("DO_ON_NEXT RUNTIME EXCEPTION")                    
                    else -> println("DO_ON_NEXT OTHER EXCEPTION")                
                }            
            }
            .subscribe(
                { println("FLOW FINISHED") },
                // 错误回调中主动抛出异常
                { throw it }
            )
    } catch (e: Exception){        
        println("CAUGHT EXCEPTION IN TRY CATCH CLAUSE")    
    }
}

注意事项 #

如果你的流通过 publishOn/subscribeOn 切换到了异步线程池执行，那么异常会在异步线程抛出，外层 try-catch 无法捕获。这种场景建议优先使用 Reactor 内置的错误处理操作符（比如 onErrorResume、onErrorReturn）处理异常，或者仍然使用 block() 统一在订阅线程接收异常。

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