使用concatDelayError时onNext/onCompleted提前回调，如何判断全部处理完成？

看起来你在使用RxJava的concatDelayError时遇到了事件提前触发的困惑，还想确认所有Observable处理完毕的时机。我来帮你一步步梳理问题并给出可行的解决方案。

问题根源分析 #

先看你代码里几个可能导致异常的点：

1. 变量名与重复定义错误：你创建的Observable是obx和oby，但后续拼接时用的是ob1和ob2，会导致编译错误；另外Observable内部的int x = doSomeThing();和方法外层的int x = 10;重名，也需要修正。
2. 冗余的ReplaySubject中转：你已经用replay().autoConnect()实现了共享订阅和事件缓存，再把结果订阅到mReplaySubject是多余的，这可能导致事件被提前转发，让你误以为是处理未完成就触发了回调。
3. 线程调度的潜在问题：如果applySchedulers()没有正确配置线程（比如未指定subscribeOn到IO线程，或observeOn到主线程），可能导致同步执行时事件立刻触发，或异步执行时回调时机混乱。

修正后的代码示例 #

先给你修正后的完整代码，再逐一解释细节：

public Observable<Integer> concatTasks() {
    Observable<Integer> ob1 = Observable.create(emitter -> {
        try {
            int resultX = doSomeThing(); // 避免和外层变量重名
            emitter.onNext(resultX);
            emitter.onComplete(); // RxJava 2+ 方法名改为onComplete()
        } catch (SQLiteException e) {
            emitter.onError(e);
        }
    }, Emitter.BackpressureMode.BUFFER);

    Observable<Integer> ob2 = Observable.create(emitter -> {
        try {
            int resultY = doSomeThing();
            emitter.onNext(resultY);
            emitter.onComplete();
        } catch (SQLiteException e) {
            emitter.onError(e);
        }
    }, Emitter.BackpressureMode.BUFFER);

    return Observable.concatDelayError(ob1, ob2)
            .compose(applySchedulers())
            .replay()
            .autoConnect();
}

// 推荐的线程调度实现（如果之前的applySchedulers有问题）
private <T> ObservableTransformer<T, T> applySchedulers() {
    return observable -> observable
            .subscribeOn(Schedulers.io()) // 让耗时操作在IO线程执行
            .observeOn(AndroidSchedulers.mainThread()); // 让回调在主线程触发
}

订阅部分去掉多余的ReplaySubject中转，直接订阅返回的Observable即可：

concatTasks().subscribe(new Observer<Integer>() {
    @Override
    public void onSubscribe(Disposable d) {
        // 可在这里保存Disposable，用于后续取消订阅
    }

    @Override
    public void onNext(Integer value) {
        // 会按顺序收到ob1、ob2发射的结果
    }

    @Override
    public void onError(Throwable e) {
        e.printStackTrace();
        // 所有Observable处理完毕且至少有一个出错时触发
        // 若多个错误，e会是CompositeException，包含所有错误信息
    }

    @Override
    public void onComplete() {
        // 所有Observable成功完成（或出错但被concatDelayError延迟处理完毕）时触发
        launchActivity(SplashActivity.this, HomeActivity.class);
        SplashActivity.this.finish();
    }
});

如何判断所有处理完成 #

concatDelayError的核心特性就是顺序执行所有Observable，即使前面的Observable出错，也会继续执行后续任务，最后统一处理错误：

• 如果所有Observable都成功完成，onComplete会在最后一个Observable的onComplete触发后被调用，这就代表所有处理都完成了。
• 如果有一个或多个Observable出错，concatDelayError会先执行完所有剩余Observable，再触发onError，此时也代表所有处理已完成（只是存在错误）。

为什么之前会出现"提前触发"的假象 #

1. 同步操作的缓存特性：如果doSomeThing()是同步操作，replay().autoConnect()会立刻订阅源Observable并执行，事件会被缓存，后续订阅时会立刻收到缓存的onNext和onComplete，这是replay的正常行为，并非提前触发。若需要延迟执行，可将autoConnect()改为autoConnect(1)，只有当第一个订阅者出现时才执行源Observable。
2. ReplaySubject的冗余中转：把replay()的结果订阅到ReplaySubject，相当于重复缓存事件，可能导致事件被重复转发，让你误以为处理未完成就触发了回调。

额外注意事项 #

• 若需要共享订阅，replay().autoConnect()已经足够，无需额外使用ReplaySubject。
• 确保doSomeThing()是异步操作，或通过subscribeOn放到IO线程，避免阻塞主线程。
• RxJava 2+版本中，onCompleted()已改名为onComplete()，注意方法名的正确性。

内容的提问来源于stack exchange，提问作者krupal.agile