无需抛出异常中断Rx-Java后续流水线的最优方案咨询

我想了解当流水线某步骤出现非异常类问题时，如何跳过后续流水线的执行。假设我有如下Kotlin编写的长流水线，希望在fooAsync返回异常情况（非抛出异常）时，避免调用开销较高的barAsync资源。

fun longAsyncCalculation(): Mono<CalculationResponse> {
    return Mono.just(Context())
       .flatMap { context ->
           fooAsync(context)
       }.flatMap { context ->
           barAsync(context)
       }.map { context ->
           createResponse(context)
       }
}

我可以通过抛出异常实现，但希望避免用异常控制程序流程；也不想在每个步骤中检查context状态，如下所示：

...
.flatMap { context ->
  if(context.hasError) {
    Mono.just(...)
  } else {
    doTheRealAsyncCall()
  }
}.flatMap { context -> 
...

请问是否存在无需抛出异常即可进入错误路径的方法？因Rx错误处理方法通常基于Throwable，我猜测可能无法实现，望提供相关最佳实践。

1. 利用filterWhen实现流短路 #

Reactor中，当流进入**完成状态（onComplete）**时，后续的flatMap等操作会自动跳过。可以使用filterWhen操作符在fooAsync执行后过滤掉带有错误的context，让流提前完成，从而跳过barAsync的调用，最后通过switchIfEmpty兜底处理错误响应。

示例代码：

fun longAsyncCalculation(): Mono<CalculationResponse> {
    return Mono.just(Context())
        .flatMap { fooAsync(it) }
        // 若context存在错误，过滤后流进入完成状态
        .filterWhen { context ->
            Mono.just(!context.hasError)
        }
        // 仅当流未提前完成时，才执行barAsync
        .flatMap { barAsync(it) }
        // 处理流提前完成的情况（即fooAsync返回错误context）
        .switchIfEmpty(Mono.just(Context().apply { markAsError() }))
        .map { createResponse(it) }
}

2. 自定义操作符封装状态检查逻辑 #

如果流水线包含多个需要状态检查的步骤，自定义通用操作符可以避免重复代码，让流水线更简洁。

示例代码：

// 自定义操作符：过滤带有错误的context，无错误则继续传递
fun <T : Context> Mono<T>.skipOnErrorState(): Mono<T> {
    return this.filter { !it.hasError }
}

fun longAsyncCalculation(): Mono<CalculationResponse> {
    return Mono.just(Context())
        .flatMap { fooAsync(it) }
        .skipOnErrorState()
        .flatMap { barAsync(it) }
        .skipOnErrorState() // 后续步骤可直接复用该操作符
        .switchIfEmpty(Mono.just(Context().apply { markAsError() }))
        .map { createResponse(it) }
}

3. 用Maybe替代Mono实现短路 #

Maybe类型支持成功、失败、完成三种状态，可以将fooAsync的返回类型改为Maybe<Context>：当检测到错误时返回Maybe.empty()，后续flatMap会自动跳过，最后通过switchIfEmpty处理空值场景。

示例代码：

// 修改fooAsync返回Maybe，错误场景返回empty
fun fooAsync(context: Context): Maybe<Context> {
    val result = doFooLogic(context)
    return if (result.hasError) Maybe.empty() else Maybe.just(result)
}

fun longAsyncCalculation(): Mono<CalculationResponse> {
    return Mono.just(Context())
        .flatMapMaybe { fooAsync(it) }
        .flatMap { barAsync(it) }
        .switchIfEmpty(Mono.just(Context().apply { markAsError() }))
        .map { createResponse(it) }
}

核心逻辑说明 #

以上方案的核心是利用Reactor流的完成状态实现短路，完全避开了异常控制流程：

• 把“非异常错误”转化为流的完成状态，让后续依赖该流的操作自动跳过
• 通过switchIfEmpty或defaultIfEmpty统一处理提前完成的错误场景，无需在每个步骤中重复检查状态

内容的提问来源于stack exchange，提问作者balint.steinbach