问题描述

最近在使用RxJava时，出现了一个错误，错误日志如下：

02-28 09:42:16.962 16647-17731/com.aspire.mpus E/AndroidRuntime: FATAL EXCEPTION: RxCachedThreadScheduler-1
                                                                Process: com.aspire.mpus, PID: 16647
                                                                java.lang.Throwable: 网络连接超时
                                                                    at com.aspire.mpus.core.api.Api.onError(Api.java:84)
                                                                    at com.aspire.mpus.core.api.Api.access$100(Api.java:23)
                                                                    at com.aspire.mpus.core.api.Api$1.subscribe(Api.java:116)
                                                                    at io.reactivex.internal.operators.observable.ObservableCreate.subscribeActual(ObservableCreate.java:40)
                                                                    at io.reactivex.Observable.subscribe(Observable.java:10514)
                                                                    at io.reactivex.internal.operators.observable.ObservableSubscribeOn$1.run(ObservableSubscribeOn.java:39)
                                                                    at io.reactivex.Scheduler$1.run(Scheduler.java:134)
                                                                    at io.reactivex.internal.schedulers.ScheduledRunnable.run(ScheduledRunnable.java:59)
                                                                    at io.reactivex.internal.schedulers.ScheduledRunnable.call(ScheduledRunnable.java:51)
                                                                    at java.util.concurrent.FutureTask.run(FutureTask.java:237)
                                                                    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:269)
                                                                    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1113)
                                                                    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:588)
                                                                    at java.lang.Thread.run(Thread.java:818)

抛出异常的原因是我在onError方法中会指定一些错误的原因，例如网络连接超时、连接服务器失败等。

 private void onError(Exception e, ObservableEmitter emitter, String message) {
        e.printStackTrace();
        LogUtils.error(e);

        if (e instanceof SocketTimeoutException) {
            emitter.onError(new Throwable("网络连接超时"));
        } else if (e instanceof ConnectException) {
            emitter.onError(new Throwable("连接服务器失败"));
        } else {
            emitter.onError(new Throwable(message));
        }
        emitter.onComplete();
    }

解决方法

判断emitter是否被取消订阅。如果当前的emitter未被取消订阅，则正常抛出异常，否则不执行。

private void onError(Exception e, ObservableEmitter emitter, String message) {
        e.printStackTrace();
        LogUtils.error(e);

        if (!emitter.isDisposed()) {
            if (e instanceof SocketTimeoutException) {
                emitter.onError(new Throwable("网络连接超时"));
            } else if (e instanceof ConnectException) {
                emitter.onError(new Throwable("连接服务器失败"));
            } else {
                emitter.onError(new Throwable(message));

            }
            emitter.onComplete();
        }
    }

重现步骤

• 开始一个网络请求
• 在请求未结束时结束当前页面
• 请求结束后，如果未能请求成功，则onError方法中会判断异常信息，并抛出异常，但是Observable未能捕捉到异常，最终导致APP crash

Observable生命周期管理

在执行网络请求时，会生成一个Disposable对象，该对象保存了Observable的状态（订阅、取消订阅）。
查看ObservableEmitter源码，

public interface ObservableEmitter<T> extends Emitter<T> {

    /**
     * Sets a Disposable on this emitter; any previous Disposable
     * or Cancellation will be unsubscribed/cancelled.
     * @param d the disposable, null is allowed
     */
    void setDisposable(@Nullable Disposable d);

    /**
     * Sets a Cancellable on this emitter; any previous Disposable
     * or Cancellation will be unsubscribed/cancelled.
     * @param c the cancellable resource, null is allowed
     */
    void setCancellable(@Nullable Cancellable c);

    /**
     * Returns true if the downstream disposed the sequence.
     * @return true if the downstream disposed the sequence
     */
    boolean isDisposed();

    /**
     * Ensures that calls to onNext, onError and onComplete are properly serialized.
     * @return the serialized ObservableEmitter
     */
    @NonNull
    ObservableEmitter<T> serialize();
}

发现ObservableEmitter也持有Disposable的属性，所以增加对当前订阅事件的判断，然后选择是否抛出异常，即可解决问题。