Mono.subscribeOn()错误处理异常问题及后台任务最佳实践咨询

一、现象分析与复现方案 #

你遇到的doOnError执行但onErrorResume不触发的情况，并非Reactor框架的异常，而是业务代码或第三方SDK异步行为导致的边界场景——核心原因是错误没有沿着Reactor的链式调用链路传递，而是脱离了订阅上下文。

可复现的模拟实现 #

要复现这个现象，只需要让错误在Reactor订阅链路之外抛出，比如模拟Azure SDK用独立线程执行操作，抛出的错误未被Reactor链路捕获：

public Mono<Void> executeBulkOperations() {
    return Mono.fromRunnable(() -> {
        // 模拟SDK内部开启独立线程执行任务
        new Thread(() -> {
            try {
                Thread.sleep(100);
                throw new RuntimeException("批量操作失败");
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }).start();
    });
}

调用这个方法时，错误在独立线程中抛出，未被Mono的链路包裹，此时doOnError会因为订阅后的副作用感知到错误，但onErrorResume作为链路内的恢复逻辑，自然无法拦截这个“脱轨”的错误。

另外一种可能是SDK内部已经捕获了异常，却通过非Reactor的方式通知错误（比如后台抛出未捕获异常但Mono已返回完成状态），这也会导致同样的现象。

二、后台即发即弃任务的错误处理最佳实践 #

针对这种后台任务，要从链路管控、错误分层处理、监控兜底三个维度优化：

1. 把所有逻辑纳入Reactor链路 #

别让第三方SDK的异步操作脱离Reactor上下文，对非Reactor友好的SDK，用Mono.fromFuture或Reactor调度器包裹，确保错误能被链路捕获：

public Mono<Void> safeExecuteBulkOperations() {
    return Mono.fromFuture(() -> CompletableFuture.runAsync(() -> {
        // 原SDK的批量操作逻辑
        throw new RuntimeException("批量操作失败");
    }, Schedulers.boundedElastic().asExecutor()))
    .subscribeOn(Schedulers.boundedElastic());
}

所有业务逻辑都要放在Mono的链式调用里，别在subscribe()之后再执行异步操作。

2. 分层处理错误，别混淆副作用与恢复逻辑 #

• doOnError仅用于日志记录、告警触发这类副作用操作，不要用它做错误恢复；
• onErrorResume/onErrorReturn才是链路内的错误恢复工具，只有错误沿着链路传递时才会触发；
• 如果遇到SDK内部的未捕获异常（错误脱离链路），可以注册Reactor全局错误处理器兜底：

  Hooks.onErrorDropped(error -> {
      log.error("后台任务出现未捕获异常", error);
      // 此处可添加告警逻辑
  });
  

3. 给即发即弃任务加监控和兜底 #

• 给每个后台任务添加唯一标识，日志中记录任务ID、参数等信息，方便排查偶发问题；
• 开启Azure SDK的日志（调整azure-core日志级别到DEBUG），查看错误发生时的调用栈，确认错误来源；
• 不要完全“即发即弃”，添加超时控制，防止任务无限阻塞：

  mono.subscribeOn(Schedulers.boundedElastic())
      .timeout(Duration.ofSeconds(30))
      .doOnError(TimeoutException.class, e -> log.warn("后台任务超时"))
      .onErrorResume(e -> {
          log.error("后台任务执行失败", e);
          return Mono.empty();
      })
      .subscribe();
  

4. 合理使用调度器，避免滥用boundedElastic #

boundedElastic适合处理阻塞IO操作，但任务过多容易耗尽线程池资源。可以自定义线程池控制并发数：

private static final Scheduler CUSTOM_BACKGROUND_POOL = Schedulers.newBoundedElastic(10, 100, "后台任务池");

// 使用自定义调度器
mono.subscribeOn(CUSTOM_BACKGROUND_POOL)
    .subscribe();

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