RxJava2并行流中onError引发UndeliverableException，如何检查下游取消状态？

这个问题在RxJava并行流开发中确实很容易踩坑——当某个并行子流发送onError后，下游的订阅会被立即终止，但其他子流的emitter.isCancelled()可能因为异步通知的延迟，暂时还返回false，如果继续发送事件就会触发UndeliverableException。下面给你几个实用的解决方案：

1. 每次发送事件前主动检查Emitter状态 #

这是最直接的预防手段，在循环或每次发送事件前，先判断emitter.isCancelled()，如果已取消就立即停止发送逻辑。虽然这个检查是即时性的，可能存在极短的异步延迟窗口，但覆盖了绝大多数常规场景。

修改你的示例代码如下：

Disposable disposable = Flowable.create(new FlowableOnSubscribe<Integer>() {
    @Override
    public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
        System.out.println("Flowable.create-emitter.isCancelled:" + emitter.isCancelled());
        for (int i = 1; i < 10; i++) {
            // 每次发送前先检查是否已取消
            if (emitter.isCancelled()) {
                System.out.println("Emitter已取消，停止发送");
                break;
            }
            if (i == 3) {
                // 模拟某个子流触发错误
                emitter.onError(new RuntimeException("Test error"));
                // 发送错误后必须终止后续发送
                break;
            }
            emitter.onNext(i);
            Thread.sleep(100); // 模拟耗时业务操作
        }
        // 确保只有在未取消且未发送错误的情况下才调用onComplete
        if (!emitter.isCancelled()) {
            emitter.onComplete();
        }
    }
})
// 并行处理，指定IO线程池
.flatMap(item -> Flowable.just(item).subscribeOn(Schedulers.io()))
.subscribe(
    item -> System.out.println("onNext: " + item),
    error -> System.err.println("onError: " + error.getMessage())
);

2. 注册取消回调，使用本地状态标记 #

因为emitter.isCancelled()的通知是异步的，我们可以给Emitter注册一个取消回调，在回调里设置一个本地的原子布尔值来标记取消状态，后续发送事件时优先检查这个本地标记，响应速度会更快。

示例代码：

Disposable disposable = Flowable.create(new FlowableOnSubscribe<Integer>() {
    @Override
    public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
        // 用原子类保证线程安全的状态标记
        AtomicBoolean isCancelled = new AtomicBoolean(false);
        
        // 注册取消回调，下游取消时立即更新本地标记
        emitter.setCancellable(() -> {
            isCancelled.set(true);
            System.out.println("当前子流已被取消");
        });

        System.out.println("Flowable.create-emitter.isCancelled:" + emitter.isCancelled());
        for (int i = 1; i < 10; i++) {
            // 优先检查本地标记，再兜底检查emitter状态
            if (isCancelled.get() || emitter.isCancelled()) {
                break;
            }
            if (i == 3) {
                emitter.onError(new RuntimeException("Test error"));
                break;
            }
            emitter.onNext(i);
            Thread.sleep(100);
        }
        if (!isCancelled.get() && !emitter.isCancelled()) {
            emitter.onComplete();
        }
    }
})
.flatMap(item -> Flowable.just(item).subscribeOn(Schedulers.io()))
.subscribe(
    item -> System.out.println("onNext: " + item),
    error -> System.err.println("onError: " + error.getMessage())
);

3. 全局兜底处理UndeliverableException #

如果上面的预防手段还是有遗漏（比如极端异步场景下的延迟），可以注册一个全局的错误处理器，捕获并处理UndeliverableException，避免程序崩溃。不过这只是兜底方案，建议优先从源头预防，而不是只处理异常。

// 在应用初始化时注册全局错误处理器
RxJavaPlugins.setErrorHandler(throwable -> {
    if (throwable instanceof UndeliverableException) {
        // 这里可以记录日志，或者根据业务需求处理
        System.err.println("捕获到UndeliverableException: " + throwable.getCause().getMessage());
        return;
    }
    // 其他未预期的异常，交给线程的默认异常处理器处理
    Thread.currentThread().getUncaughtExceptionHandler()
          .uncaughtException(Thread.currentThread(), throwable);
});

额外注意事项 #

• 发送onError后，必须立即终止后续的事件发送逻辑，因为RxJava规范中onError是终止事件，下游不会再处理任何后续的onNext或onComplete。
• 在使用flatMap实现并行时，可以通过flatMap(item -> ..., maxConcurrency)指定最大并行数，避免过多子流同时运行，减少取消延迟的概率。

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