RxSwift中替代blockingGet的生产环境可用方案咨询

一、放弃Blocking调用，改用适配旧线程的非阻塞订阅 #

生产环境使用blockingGet()既违背RxJava的异步设计初衷，还容易引发死锁、线程阻塞等问题，尤其旧代码的线程模型通常未考虑Rx的调度逻辑。正确的做法是将Observable的结果回调到旧线程：

• 用observeOn()指定旧代码的线程（如果旧线程由Handler管理，可使用HandlerScheduler或自定义Scheduler）
• 在旧代码中持有Disposable引用，避免订阅被提前回收，同时在旧组件生命周期结束时（如页面销毁、对象释放）取消订阅

示例代码：

// 旧线程绑定的Handler，假设旧代码用它控制线程执行
Handler oldThreadHandler = ...;

// 调用新仓库的Observable
Disposable disposable = newRepository.getDataObservable()
    .subscribeOn(Schedulers.io()) // 指定仓库数据源的工作线程
    .observeOn(HandlerScheduler.from(oldThreadHandler)) // 切换到旧线程处理结果
    .first()
    .subscribe(
        data -> {
            // 在旧线程中直接调用原有业务逻辑处理数据
            oldCodeProcessData(data);
        },
        error -> {
            // 回调旧代码的错误处理逻辑
            oldCodeHandleError(error);
        }
    );

// 在旧组件销毁时（如Activity的onDestroy、自定义对象的release方法）取消订阅
// 避免内存泄漏和野指针问题
oldComponent.saveDisposable(disposable);

二、解决EXC_BAD_ACCESS崩溃问题 #

该崩溃基本由内存管理不当导致，常见原因及修复方式：

• 未持有Disposable导致订阅被GC回收：旧代码对象生命周期较短，若未保留Disposable引用，订阅会被提前回收，回调时访问已释放对象。必须在旧组件中持有Disposable，并在生命周期结束时调用dispose()。
• first()的空发射异常：若Observable未发射任何数据就完成，first()会抛出NoSuchElementException，未处理该异常可能引发崩溃。可改用firstOrError()明确处理无数据场景，或first(defaultValue)提供默认值：

// 改用firstOrError，强制处理无数据的错误情况
newRepository.getDataObservable()
    .firstOrError()
    .subscribe(...)

• 旧线程的线程安全问题：旧线程可能缺乏内存屏障，导致Rx回调时访问的对象处于不一致状态。确保旧线程处理数据时，所有对象访问都是线程安全的，或通过subscribeOn/observeOn严格控制线程切换逻辑。
• 回调对象提前释放：订阅时传入的Consumer等回调对象若被提前释放，会触发EXC_BAD_ACCESS。可配合弱引用和Disposable销毁逻辑，或保证回调对象的生命周期长于订阅周期。

三、封装兼容旧线程的工具类 #

如果旧代码大量需要调用新仓库，可封装工具类统一处理线程切换和订阅管理：

public class RxOldThreadAdapter {
    private final Handler oldThreadHandler;

    public RxOldThreadAdapter(Handler oldThreadHandler) {
        this.oldThreadHandler = oldThreadHandler;
    }

    public <T> Disposable execute(Observable<T> observable, Consumer<T> onSuccess, Consumer<Throwable> onError) {
        return observable
            .subscribeOn(Schedulers.io())
            .observeOn(HandlerScheduler.from(oldThreadHandler))
            .firstOrError()
            .subscribe(onSuccess, onError);
    }
}

// 旧代码中调用示例
RxOldThreadAdapter adapter = new RxOldThreadAdapter(oldThreadHandler);
Disposable disposable = adapter.execute(
    newRepository.getDataObservable(),
    data -> oldCodeProcessData(data),
    error -> oldCodeHandleError(error)
);
// 将disposable存入旧组件的管理列表，销毁时统一清理

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