189 8069 5689
这篇文章给大家介绍RxJava2中是如何对异常进行处理的，内容非常详细，感兴趣的小伙伴们可以参考借鉴，希望对大家能有所帮助。
成都创新互联公司是由多位在大型网络公司、广告设计公司的优秀设计人员和策划人员组成的一个具有丰富经验的团队，其中包括网站策划、网页美工、网站程序员、网页设计师、平面广告设计师、网络营销人员及形象策划。承接：成都做网站、网站制作、网站改版、网页设计制作、网站建设与维护、网络推广、数据库开发,以高性价比制作企业网站、行业门户平台等全方位的服务。
众所周知，RxJava2 中当链式调用中抛出异常时，如果没有对应的 Consumer 去处理异常，则这个异常会被抛出到虚拟机中去，Android 上的直接表现就是 crash，程序崩溃。
订阅方式
说异常处理前咱们先来看一下 RxJava2 中 Observable 订阅方法 subscribe() 我们常用的几种订阅方式：
// 1 subscribe() // 2 Disposable subscribe(Consumer onNext) // 3 Disposable subscribe(Consumer onNext, Consumer onError) // 4 Disposable subscribe(Consumer onNext, Consumer onError,Action onComplete) // 5 Disposable subscribe(Consumer onNext, Consumer onError,Action onComplete, Consumer onSubscribe) // 6 void subscribe(Observer observer)
无参和以 Consumer为参数的几种方法内部都是以默认参数补齐的方式最终调用第 5 个方法，而方法 5 内部通过 LambdaObserver 将参数包装成 Observer 再调用第 6 个方法
public final Disposable subscribe(Consumer onNext, Consumer onError,
Action onComplete, Consumer 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 ls = new LambdaObserver(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> {
override fun onComplete() {
}
override fun onSubscribe(d: Disposable) {
}
override fun onNext(t: List) {
throw RuntimeException("runtime exception")
}
override fun onError(e: Throwable) {
Log.d("error", e.message)
}
}) 结果：不会触发 onError，App 崩溃
2、Observer onNext 中抛出异常（未切换线程）
Observable.create{ it.onNext("ssss") } .subscribe(object : Observer { 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> {
override fun onComplete() {
}
override fun onSubscribe(d: Disposable) {
}
override fun onNext(t: List) {
}
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 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{ 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