RxJava中ObservableOnSubscribe抛异常致应用崩溃问题咨询

我来帮你分析下这个问题的根源，以及怎么解决它：

问题核心原因 #

你现在的写法有两个关键问题，导致了偶尔的崩溃：

1. 错误传递方式不符合RxJava规范：在ObservableOnSubscribe.subscribe()方法里直接抛出异常，不是RxJava推荐的错误传递方式。RxJava设计时就要求我们通过ObservableEmitter.onError(Throwable)来向下游传递错误，而不是直接throw。
2. 线程竞态条件：即使你判断了!oe.isDisposed()，也可能出现"检查时还没取消，抛异常时刚好取消"的线程竞态。此时RxJava尝试把异常传递给观察者，但观察者已经没有处理错误的回调了，就会抛出OnErrorNotImplementedException。

另外，直接抛异常还会绕过你后面compose、map里的错误处理逻辑，让异常直接扩散到RxJava的内部线程，引发崩溃。

修复方案 #

把直接抛出异常的逻辑改成调用emitter.onError()，这是RxJava处理错误的标准姿势。同时要注意，发送onError后就不要再发送onComplete了——RxJava要求一个Observable只能发送一个终止事件（要么onError，要么onComplete）。

修改后的代码示例 #

原来的代码：

public void subscribe(ObservableEmitter oe) throws Exception {
    try{
        // 你的网络请求等业务逻辑
    } catch (Exception ex) {
        if (!oe.isDisposed()) {
            if (ex instanceof IOException) {
                throw new NetException(...); // 这里是崩溃的根源
            }
        } else{
            // 处理已取消订阅的情况
        }
    }
    oe.onComplete();
}

修改后：

public void subscribe(ObservableEmitter oe) throws Exception {
    try{
        // 你的网络请求等业务逻辑
        
        // 只有正常完成时才调用onComplete
        oe.onComplete();
    } catch (Exception ex) {
        if (!oe.isDisposed()) {
            Throwable error = ex instanceof IOException ? new NetException(...) : ex;
            oe.onError(error); // 用onError传递错误
        }
        // 这里不要再调用onComplete！因为onError已经是终止事件了
    }
}

额外的安全保障 #

另外，在调用subscribe()的时候，一定要实现onError回调，不要只传onNext：

Observable.create(new ObservableOnSubscribe())
    .subscribeOn(Schedulers.io())
    .compose(new ParserTransformer())
    .map(...)
    .subscribe(
        result -> { /* 处理成功结果 */ },
        error -> { /* 处理错误，比如弹出提示、记录日志 */ }
    );

这样即使有错误流下来，也会被你的onError回调处理，不会触发OnErrorNotImplementedException。

为什么这样能解决问题？ #

RxJava的onError()方法会严格遵守订阅状态：如果订阅已经取消，它会静默处理这个错误，不会尝试传递给观察者。同时，错误会正常经过你上游的compose、map等操作符，让你能在合适的位置统一处理错误，而不是让异常直接"逃逸"引发崩溃。

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