RxJava2中是如何對異常進行處理的

這篇文章給大家介紹RxJava2中是如何對異常進行處理的，內容非常詳細，感興趣的小伙伴們可以參考借鑒，希望對大家能有所幫助。

眾所周知，RxJava2 中當鏈式調用中拋出異常時，如果沒有對應的 Consumer 去處理異常，則這個異常會被拋出到虛擬機中去，Android 上的直接表現就是 crash，程序崩潰。

訂閱方式

說異常處理前咱們先來看一下 RxJava2 中 Observable 訂閱方法 subscribe() 我們常用的幾種訂閱方式：

// 1
subscribe()
// 2
Disposable subscribe(Consumer<? super T> onNext)
// 3
Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError)
// 4
Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError,Action onComplete)
// 5
Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError,Action onComplete, Consumer<? super Disposable> onSubscribe)
// 6
void subscribe(Observer<? super T> observer)

無參和以 Consumer為參數的幾種方法內部都是以默認參數補齊的方式最終調用第 5 個方法，而方法 5 內部通過 LambdaObserver 將參數包裝成 Observer 再調用第 6 個方法

 public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError,
 Action onComplete, Consumer<? super Disposable> onSubscribe) {
 ObjectHelper.requireNonNull(onNext, "onNext is null");
 ObjectHelper.requireNonNull(onError, "onError is null");
 ObjectHelper.requireNonNull(onComplete, "onComplete is null");
 ObjectHelper.requireNonNull(onSubscribe, "onSubscribe is null");

 LambdaObserver<T> ls = new LambdaObserver<T>(onNext, onError, onComplete, onSubscribe);

 subscribe(ls);

 return ls;
 }

所以使用 Consumer 參數方式和  Observer 參數方式進行訂閱除了觀察回調來源不一樣其他沒有任何差別。但就是因為這種差別，在異常情況發生時的處理結果上也會產生差別

異常處理

我們分別進行一下幾種方式模擬異常：

1、Observer onNext 中拋出異常（切換線程）

 apiService.newJsonKeyData()
  .doOnSubscribe { t -> compositeDisposable.add(t) }
  .compose(RxScheduler.sync()) // 封裝的線程切換
  .subscribe(object : Observer<List<ZooData>> {
  override fun onComplete() {

  }

  override fun onSubscribe(d: Disposable) {

  }

  override fun onNext(t: List<ZooData>) {
  throw RuntimeException("runtime exception")
  }

  override fun onError(e: Throwable) {
  Log.d("error", e.message)
  }

  })

結果：不會觸發 onError，App 崩潰

2、Observer onNext 中拋出異常（未切換線程）

  Observable.create<String> {
   it.onNext("ssss")
   }
    .subscribe(object : Observer<String> {
    override fun onComplete() {

    }

    override fun onSubscribe(d: Disposable) {

    }

    override fun onNext(t: String) {
     Log.d("result::", t)
     throw RuntimeException("run llllll")
    }

    override fun onError(e: Throwable) {
     Log.e("sss", "sss", e)
    }

    })

結果：會觸發 onError，App 未崩潰

3、Observer map 操作符中拋出異常

  apiService.newJsonKeyData()
   .doOnSubscribe { t -> compositeDisposable.add(t) }
   .map {
   throw RuntimeException("runtime exception")
   }
   .compose(RxScheduler.sync())
   .subscribe(object : Observer<List<ZooData>> {
   override fun onComplete() {

   }

   override fun onSubscribe(d: Disposable) {

   }

   override fun onNext(t: List<ZooData>) {

   }

   override fun onError(e: Throwable) {
    Log.d("error", e.message)
   }

   })

結果：會觸發 Observer 的 onError，App 未崩潰

4、Consumer onNext 中拋出異常

  apiService.newJsonKeyData()
   .doOnSubscribe { t -> compositeDisposable.add(t) }
   .compose(RxScheduler.sync())
   .subscribe({
   throw RuntimeException("messsasassssssssssssssssssssssssssssssssssssss")
   }, {
   Log.d("Error", it.message)
   })

結果 A：有 errorConsumer 觸發 errorConsumer，App 未崩潰

 apiService.newJsonKeyData()
   .doOnSubscribe { t -> compositeDisposable.add(t) }
   .compose(RxScheduler.sync())
   .subscribe {
   throw RuntimeException("messsasassssssssssssssssssssssssssssssssssssss")
   }

結果 B：無 errorConsumer，App 崩潰

那么為什么會出現這些不同情況呢？我們從源碼中去一探究竟。

Consumer 訂閱方式的崩潰與不崩潰

subscribe() 傳入 consumer 類型參數最終在 Observable 中會將傳入的參數轉換為 LambdaObserver 再調用 subscribe(lambdaObserver)進行訂閱。展開  LambdaObserver：(主要看 onNext 和 onError 方法中的處理)

		.
		.
		.
		 @Override
 public void onNext(T t) {
 if (!isDisposed()) {
  try {
  onNext.accept(t);
  } catch (Throwable e) {
  Exceptions.throwIfFatal(e);
  get().dispose();
  onError(e);
  }
 }
 }

 @Override
 public void onError(Throwable t) {
 if (!isDisposed()) {
  lazySet(DisposableHelper.DISPOSED);
  try {
  onError.accept(t);
  } catch (Throwable e) {
  Exceptions.throwIfFatal(e);
  RxJavaPlugins.onError(new CompositeException(t, e));
  }
 } else {
  RxJavaPlugins.onError(t);
 }
 }
		.
		.
		.

onNext 中調用了對應 consumer 的 apply() 方法，并且進行了 try catch。因此我們在 consumer 中進行的工作拋出異常會被捕獲觸發 LambdaObserver 的 onError。再看 onError 中，如果訂閱未取消且 errorConsumer 的 apply() 執行無異常則能正常走完事件流，否則會調用 RxJavaPlugins.onError(t)。看到這里應該就能明白了，當訂閱時未傳入 errorConsumer時 Observable 會指定 OnErrorMissingConsumer 為默認的 errorConsumer，發生異常時拋出 OnErrorNotImplementedException。

RxJavaPlugins.onError(t)

上面分析，發現異常最終會流向 RxJavaPlugins.onError(t)。這個方法為 RxJava2 提供的一個全局的靜態方法。

 public static void onError(@NonNull Throwable error) {
 Consumer<? super Throwable> f = errorHandler;

 if (error == null) {
  error = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
 } else {
  if (!isBug(error)) {
  error = new UndeliverableException(error);
  }
 }

 if (f != null) {
  try {
  f.accept(error);
  return;
  } catch (Throwable e) {
  // Exceptions.throwIfFatal(e); TODO decide
  e.printStackTrace(); // NOPMD
  uncaught(e);
  }
 }

 error.printStackTrace(); // NOPMD
 uncaught(error);
 }

查看其源碼發現，當 errorHandler 不為空時異常將由其消耗掉，為空或者消耗過程產生新的異常則 RxJava 會將異常拋給虛擬機（可能導致程序崩潰）。 errorHandler本身是一個 Consumer 對象，我們可以通過如下方式配置他：

 RxJavaPlugins.setErrorHandler(object : Consumer1<Throwable> {
 override fun accept(t: Throwable?) {
  TODO("not implemented") //To change body of created functions use File | Settings | File Templates.
 }

 })

數據操作符中拋出異常

以 map 操作符為例，map 操作符實際上 RxJava 是將事件流 hook 了另一個新的 Observable ObservableMap

 @CheckReturnValue
 @SchedulerSupport(SchedulerSupport.NONE)
 public final <R> Observable<R> map(Function<? super T, ? extends R> mapper) {
 ObjectHelper.requireNonNull(mapper, "mapper is null");
 return RxJavaPlugins.onAssembly(new ObservableMap<T, R>(this, mapper));
 }

進入 ObservableMap 類，發現內部訂閱了一個內部靜態類 MapObserver，重點看 MapObserver  的 onNext 方法

 public void onNext(T t) {
  if (done) {
  return;
  }

  if (sourceMode != NONE) {
  downstream.onNext(null);
  return;
  }

  U v;

  try {
  v = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper function returned a null value.");
  } catch (Throwable ex) {
  fail(ex);
  return;
  }
  downstream.onNext(v);
 }

onNext 中 try catch 了 mapper.apply()，這個 apply 執行的就是我們在操作符中實現的 function 方法。因此在 map 之類數據變換操作符中產生異常能夠自身捕獲并發送給最終的 Observer。如果此時的訂閱對象中能消耗掉異常則事件流正常走 onError() 結束,如果訂閱方式為上以節中的 consumer，則崩潰情況為上一節中的分析結果。

Observer 的 onNext 中拋出異常

上述的方式 1 為一次網絡請求，里面涉及到線程的切換。方式 2 為直接 create 一個 Observable 對象，不涉及線程切換，其結果為線程切換后,觀察者 Observer 的 onNext() 方法中拋出異常無法觸發 onError()，程序崩潰。

未切換線程的 Observable.create

查看 create() 方法源碼，發現內部創建了一個 ObservableCreate 對象，在調用訂閱時會觸發 subscribeActual()  方法。在  subscribeActual() 中再調用我們 create 時傳入的 ObservableOnSubscribe 對象的 subscribe() 方法來觸發事件流。

 @Override
 protected void subscribeActual(Observer<? super T> observer) {
	
		// 對我們的觀察者使用 CreateEmitter 進行包裝,內部的觸發方法是相對應的
 CreateEmitter<T> parent = new CreateEmitter<T>(observer);
 observer.onSubscribe(parent);

 try {
			// source 為 create 時創建的 ObservableOnSubscribe 匿名內部接口實現類
  source.subscribe(parent);
 } catch (Throwable ex) {
  Exceptions.throwIfFatal(ex);
  parent.onError(ex);
 }
 }

上述代碼中的訂閱過程是使用 try catch 今夕包裹的。訂閱及訂閱觸發后發送的事件流都在一個線程，所以能夠捕獲整個事件流中的異常。（PS : 大家可以嘗試下使用  observeOn() 切換事件發送線程。會發現異常不能再捕獲，程序崩潰）

涉及線程變換時的異常處理

Retrofit 進行網絡請求返回的 Observable 對象實質上是 RxJava2CallAdapter 中生成的 BodyObservable,期內部的 onNext 是沒有進行異常捕獲的。其實這里是否捕獲并不是程序崩潰的根本原因，因為進行網絡請求，必然是涉及到線程切換的。就算此處 try catch 處理了，也并不能捕獲到事件流下游的異常。

 @Override public void onNext(Response<R> response) {
 if (response.isSuccessful()) {
 observer.onNext(response.body());
 } else {
 terminated = true;
 Throwable t = new HttpException(response);
 try {
  observer.onError(t);
 } catch (Throwable inner) {
  Exceptions.throwIfFatal(inner);
  RxJavaPlugins.onError(new CompositeException(t, inner));
 }
 }
 }

以我們在最終的 Observer 的 onNext 拋出異常為例，要捕獲這次異常那么必須在最終的調用線程中去進行捕獲。即 .observeOn(AndroidSchedulers.mainThread()) 切換過來的 Android 主線程。與其他操作符一樣，線程切換時產生了一組新的訂閱關系，RxJava 內部會創建一個新的觀察對象 ObservableObserveOn。

 @Override
 public void onNext(T t) {
  if (done) {
  return;
  }

  if (sourceMode != QueueDisposable.ASYNC) {
  queue.offer(t);
  }
  schedule();
 }
		.
		.
		.
		void schedule() {
  if (getAndIncrement() == 0) {
  worker.schedule(this); // 執行 ObservableObserveOn 的 run 方法
  }
 }
		.
		.
		.
	 @Override
 public void run() {
  if (outputFused) {
  drainFused();
  } else {
  drainNormal();
  }
 }	

而執行任務的 worker 即為對應線程 Scheduler 的對應實現子類所創建的 Worker，以 AndroidSchedulers.mainThread() 為例，Scheduler 實現類為 HandlerScheduler，其對應 Worker 為 HandlerWorker，最終任務交給 ScheduledRunnable 來執行。

 private static final class ScheduledRunnable implements Runnable, Disposable {
 private final Handler handler;
 private final Runnable delegate;

 private volatile boolean disposed; // Tracked solely for isDisposed().

 ScheduledRunnable(Handler handler, Runnable delegate) {
  this.handler = handler;
  this.delegate = delegate;
 }

 @Override
 public void run() {
  try {
  delegate.run();
  } catch (Throwable t) {
  RxJavaPlugins.onError(t);
  }
 }

 @Override
 public void dispose() {
  handler.removeCallbacks(this);
  disposed = true;
 }

 @Override
 public boolean isDisposed() {
  return disposed;
 }
 }

會發現，run 中 進行了 try catch。但 catch 內消化異常使用的是全局異常處理 RxJavaPlugins.onError(t);，而不是某一個觀察者的 onError。所以在經過切換線程操作符后，觀察者 onNext 中拋出的異常，onError 無法捕獲。

處理方案

既然知道了問題所在，那么處理問題的方案也就十分清晰了。

1、注冊全局的異常處理

 RxJavaPlugins.setErrorHandler(object : Consumer<Throwable> {
  override fun accept(t: Throwable?) {
  // do something 
  }

 })

2、Consumer 作為觀察者時，不完全確定沒有異常一定要添加異常處理 Consumer

 apiService.stringData()
   .doOnSubscribe { t -> compositeDisposable.add(t) }
   .compose(RxScheduler.sync())
   .subscribe(Consumer<Boolean>{ }, Consumer<Throwable> { })

3、Observer 可以創建一個 BaseObaerver 將 onNext 內部進行 try catch 人為的流轉到 onError 中，項目中的觀察這都使用這個 BaseObserver 的子類。

 @Override
 public void onNext(T t) {
 try {
  onSuccess(t);
 } catch (Exception e) {
  onError(e);
 }
 data = t;
 success = true;
 }

關于RxJava2中是如何對異常進行處理的就分享到這里了，希望以上內容可以對大家有一定的幫助，可以學到更多知識。如果覺得文章不錯，可以把它分享出去讓更多的人看到。