java

5.1 错误处理：onErrorReturn、onErrorResume、retry

系统讲解 Reactor 错误处理操作符的语义与选型：onErrorReturn（替换默认值）、onErrorResume（切换备用流）、onErrorMap（异常转换）与 retry（重试N次）。通过实跑代码展示四种策略在数据库查询失败、HTTP 超时等场景下的行为差异，重点说明 retry 的「订阅级重试」本质与无限重试的资源泄漏风险。

Leeting Yan 4 分钟阅读 1896 字

错误不是终点

在响应式流中，onError 是一个终止信号——一旦发出，整个流就结束了。但业务上，很多「错误」是可以恢复的：网络抖动可以重试、缓存命中可以降级、部分失败可以忽略。

Reactor 提供了四种错误处理策略：

操作符行为适用场景
onErrorReturn用默认值替换错误可忽略的错误
onErrorResume切换到备用流降级逻辑
onErrorMap转换异常类型异常封装
retry重新订阅并从头开始瞬态错误

onErrorReturn：用默认值替换

Flux.just(1, 2, 0, 4)
    .map(i -> 10 / i)  // i=0 时除零异常
    .onErrorReturn(-1)
    .subscribe(System.out::println);

// 输出: 10, 5, -1
// 注意：0 之后的 4 不会处理，流已终止

onErrorReturn 在错误发生时发射一个默认值，然后调用 onComplete。它不会恢复原始流，只是给了下游一个「优雅的失败」。

带条件的 onErrorReturn

Flux.just(1, 2, 0, 4)
    .map(i -> 10 / i)
    .onErrorReturn(e -> e instanceof ArithmeticException, -1)
    .subscribe(System.out::println);

// 只有 ArithmeticException 被替换为 -1，其他异常仍然传播

陷阱：onErrorReturn 会终止流

Flux.just(1, 2, 0, 4)
    .map(i -> 10 / i)
    .onErrorReturn(-1)
    .collectList()
    .subscribe(list -> System.out.println(list));

// 输出: [10, 5, -1]
// 不是 [10, 5, -1, 2]！4 被跳过了

如果你需要继续处理后续元素，onErrorReturn 不够——需要更细粒度的错误处理（如 onErrorContinue，但它是 Reactor 的扩展，不是 RS 标准）。

onErrorResume：切换到备用流

Flux.just("user1", "user2", "user3")
    .flatMap(this::fetchUserFromPrimary)
    .onErrorResume(e -> fetchUserFromBackup())
    .subscribe(System.out::println);

当主数据源失败时，自动切换到备用数据源。这是微服务架构中常见的降级策略。

多个降级层级

Flux.just("user1")
    .flatMap(this::fetchUserFromPrimary)
    .onErrorResume(e -> fetchUserFromBackup())
    .onErrorResume(e -> fetchUserFromCache())
    .onErrorReturn(User.UNKNOWN)
    .subscribe();

陷阱：备用流也会失败

// 危险：备用流也可能失败
.onErrorResume(e -> {
    // 如果这里也抛异常，流直接终止
    return dangerousFallback();
})

正确做法：备用流应该用 onErrorReturn 或 onErrorResume 兜底。

onErrorMap：转换异常类型

Flux.just(1, 2, 3)
    .map(i -> {
        if (i == 2) throw new IllegalArgumentException("bad value");
        return i;
    })
    .onErrorMap(e -> new BusinessException("处理失败", e))
    .subscribe(
        System.out::println,
        e -> System.out.println("Error: " + e.getClass().getSimpleName())
    );

// 输出: 1, Error: BusinessException

onErrorMap 不改变流的终止行为，只是把异常转换成另一种类型。这在分层架构中很有用——DAO 层的 SQLException 转换成 Service 层的 BusinessException。

retry：重新订阅

retry 不是「把当前元素重试一遍」，而是重新订阅整个上游流。

AtomicInteger attempt = new AtomicInteger(0);

Flux<Integer> flux = Flux.defer(() -> {
    int a = attempt.incrementAndGet();
    System.out.println("Attempt " + a);
    if (a < 3) {
        return Flux.error(new RuntimeException("fail " + a));
    }
    return Flux.just(1, 2, 3);
});

flux.retry(3)
    .subscribe(System.out::println);

// 输出:
// Attempt 1
// Attempt 2
// Attempt 3
// 1, 2, 3

注意：retry(3) 意味着最多重试 3 次。如果 3 次都失败，最终会把错误抛给下游。

retry 的陷阱：无限重试

// 危险！无限重试
flux.retry()
    .subscribe();

// 如果上游永远失败，这个流会无限重订阅，持续占用线程与连接

正确做法：始终限制重试次数。

带退避的 retry

Reactor 3.8 的 retryWhen 只接受 Retry 对象，最常用的是内置的 Retry.backoff：

import reactor.util.retry.Retry;

flux.retryWhen(Retry.backoff(3, Duration.ofMillis(100)))
    .subscribe();

口径是一致的：retry(3)、Retry.backoff(3, ...)、Retry.max(3) 里的 3 都表示「最多重试 3 次」，即最多 4 次订阅。实测 Retry.backoff(3, ...) 对永远失败的源会订阅 4 次，耗尽后抛 RetryExhaustedException: Retries exhausted: 3/3。

Retry.withThrowable(errors -> ...) 仍然存在，用于把 Flux<Throwable> 映射成自定义 companion，属于兼容旧代码的入口。新代码优先用 Retry.backoff / Retry.fixedDelay / Retry.max，它们对「重试耗尽后如何把原错误交给下游」有明确语义；手写 companion 如果直接让它 complete，可能把原始异常吞掉。

四种策略的选择指南

场景策略代码
某个元素处理失败，用默认值onErrorReturn.onErrorReturn(defaultValue)
主服务失败，切备用服务onErrorResume.onErrorResume(e -> fallback())
异常类型不统一，需要封装onErrorMap.onErrorMap(e -> new BizException(e))
网络抖动，可以重试retry.retry(3)
需要退避重试retryWhen.retryWhen(Retry.backoff(3, ...))；手写逻辑用 Retry.withThrowable(errors -> ...)

错误处理的顺序

操作符链中的错误处理是有顺序的：

Flux.just(1, 2, 0)
    .map(i -> 10 / i)
    .onErrorReturn(-1)           // 第一层：替换为 -1
    .onErrorResume(e -> Mono.just(-2))  // 不会执行，因为 -1 已处理
    .subscribe();

onErrorReturn 先匹配，错误被替换后流终止，onErrorResume 不会触发。

全局错误处理

除了操作符级的错误处理，还可以在订阅时处理：

flux.subscribe(
    System.out::println,           // onNext
    e -> log.error("Unhandled", e), // onError
    () -> log.info("Complete")      // onComplete
);

如果操作符链中没有错误处理，最终错误会到达 subscribe 的 onError 回调。如果没有提供 onError 消费者，Reactor 会把错误交给 Hooks.onErrorDropped 的默认处理器：打印 Operator called default onErrorDropped 及堆栈，但不会终止 JVM。生产环境应显式提供 onError 回调，或通过 Hooks.onErrorDropped 接入日志/告警。

错误处理模式总结

模式一：降级

public Mono<Order> getOrder(String id) {
    return primaryService.getOrder(id)
        .onErrorResume(e -> {
            log.warn("Primary failed, using backup", e);
            return backupService.getOrder(id);
        })
        .onErrorResume(e -> {
            log.error("Backup also failed", e);
            return cache.getOrder(id);
        })
        .onErrorReturn(Order.EMPTY);
}

模式二：重试+熔断

public Mono<Order> getOrderWithResilience(String id) {
    return circuitBreaker.run(
        () -> primaryService.getOrder(id)
            .retryWhen(Retry.backoff(3, Duration.ofMillis(200)))
            .timeout(Duration.ofSeconds(2)),
        throwable -> backupService.getOrder(id)
    );
}

模式三：部分失败容忍

public Flux<Order> getOrders(List<String> ids) {
    return Flux.fromIterable(ids)
        .flatMap(id -> getOrder(id)
            .onErrorResume(e -> {
                log.warn("Failed to get order {}", id, e);
                return Mono.empty();  // 跳过失败的
            }))
        .collectList()
        .flatMapMany(Flux::fromIterable);
}

错误日志最佳实践

场景日志级别内容
可恢复错误（重试成功）WARN错误 + 重试次数
降级成功INFO降级源 + 原始错误
最终失败ERROR完整堆栈 + 上下文
业务错误（参数非法）DEBUG不触发告警

小结

操作符行为是否终止场景
onErrorReturn替换为默认值是（发射默认值后完成）可忽略的错误
onErrorResume切换备用流是（备用流完成后）服务降级
onErrorMap转换异常类型是异常封装
retry重新订阅否（成功继续）瞬态错误
retryWhen条件+延迟重试否网络抖动

下一节，我们用 retryWhen 实现指数退避重试，并讨论与熔断的边界。

阅读导航

上一篇：4.3 组合操作符：zip、merge、combineLatest

下一篇：5.2 指数退避重试与熔断信号

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「java」更多文章

  1. 15.3 全书总结与学习路径回顾
  2. 15.2 响应式流处理的演进趋势
  3. 15.1 引擎选型指南