从最近的Rxjava2异常说起
最新的项目代码把RxJava1的代码升级到了RxJava2.发布后发现crash概率提高了许多.简单地总结就是之前RxJava1中在Subscriber中onNext方法,如果发生了异常.默认会进入Subscriber的onError.但是在RxJava2中直接抛出到了虚拟机.导致crash.由于RxJava1这种消费所有异常特性.导致在老版本code中诸多不严谨代码的bug都被rxjava1隐藏掉了.尤其在当前应用,业务与UI极度耦合的情况下,业务线程返回后UI已经被销毁或者改动.这种情况随处可见.
RxJava的重要组成Observable(被观察者) Observer(观察者在Rxjava2中最终是Consumer)
其实说RxJava2中Observer的onNext回调抛出异常会crash.是非常不严谨的说法.因为RxJava的操作符以及链式编程的调用结构.我们最终的订阅者Observer会在操作符的帮助下.层层包裹最后被调用.
Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("normal data test");
emitter.onComplete();
}
});
Observer observer = new Observer<String>() {
@Override
public void onComplete() {
System.out.println("onComplete");
}
@Override
public void onError(Throwable arg0) {
System.out.println("error = " + arg0.getMessage());
}
@Override
public void onNext(String arg0) {
System.out.println("onnext");
throw new RuntimeException(arg0);
}
@Override
public void onSubscribe(Disposable arg0) {
}
};
observable.subscribe(observer);
上边这个代码,也不会引起崩溃.onNext中throw的异常会进入onError中print.
其原因在与
Observable.create()方法,所生成的ObservableCreate对象,ObservableCreate是Observable的子类,他的subscribe方法,会包裹我们传入的observer,使用CreateEmitter(发射器)这个类,最终回调传入observer的onNext.source是我们create方法传入的ObservableOnSubscribe对象,调用他的subscribe方法,其实就是执行其内部的方法体.
emitter.onNext("normal data test");
emitter.onComplete();
这部分操作被try-catch包裹,可以看到异常也会进入emitter的onError最终给Observer
@Override
protected void subscribeActual(Observer<? super T> observer) {
CreateEmitter<T> parent = new CreateEmitter<T>(observer);
observer.onSubscribe(parent);
try {
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
}
不过对应我们项目中的Observable并不是这个简单地ObservableCreate.
Retrofit.create(IApi).subscribeOn().observeOn().map().subsribe();//其实还有个retryWhen
Retrofit.create这个方法生成的Observable实在Rxjava2CallAdapter的adapter生成的
BodyObservable
subscribeOn也会生成对应的ObservableSbuscribeOn
observeOn生成的是ObservableObserverOn
map则对应生成ObservableMap
考虑到observeOn方法控制了Rxjava的线程切换,所以,ObservableObserverOn这个实现类的内部必然有线程切换,所以他之前的Observable对象即使try了自己内部对应的observer的onNext方法,也不可能对往下的传递生效(听明白了吗?).所以,只有两个位置还有可能try住我们最终observer的onNext方法.
ObservableObserverOn
@Override
protected void subscribeActual(Observer<? super T> observer) {
if (scheduler instanceof TrampolineScheduler) {
source.subscribe(observer);
} else {
Scheduler.Worker w = scheduler.createWorker();
source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
}
}
ObserveOnObserver
@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);//把事件切换回主线程
}
}
@Override
public void run() {
if (outputFused) {
drainFused();
} else {
drainNormal();//这里开始在主线程调用
}
}
void drainNormal() {
//这里边的内容就不粘贴了占地方.这里边1.没有对下一个observable的onNext没有任何处理,没有负责try-catch
}
这里的worker其实HandlerScheduler,他被AndroidSchedulers持有,RxAndroid,负责把往后的事件切换回主线程.他的schedule方法,其实就是执行ObserveOnObserver的run方法.
其中ObserveOnObserver传递给HandlerScheduler,也会进行一层封装ScheduledRunnable,这里边会捕获异常,不过他没有把exception交给observer的onError而是交给全局plugin.
@Override
public void run() {
try {
delegate.run();
} catch (Throwable t) {
RxJavaPlugins.onError(t);//这里可以做全局的异常处理.
}
}
最终事件发射给ObservableMap
@Override
public void subscribeActual(Observer<? super U> t) {
source.subscribe(new MapObserver<T, U>(t, function));
}
@Override
public void onNext(T t) {
if (done) {
return;
}
if (sourceMode != NONE) {
actual.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;
}
actual.onNext(v);//终于到了我们传入的observer的onNext
}
这部分代码就非常简单了.最后简单地调动v = mapper.apply(t)方法,做数据的转换,然后调用我们最后传入的真正的Observer的onNext(v)方法.可以看到这里也没有处理异常情况.
所以到此为止.最终,如果我们传入的Observer的onNext方法,抛出了异常.他的catch位置就是HandlerScheduler ScheduledRunnable这个内部类的run方法.
这个Exception最终是传递给RxJavaPlugins这个对象处理.
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这个静态对象,他就可以不把这个异常.抛给虚拟机.
所以,针对这次Rxjava2的异常.最简单的方法就是给RxJavaPlugins设置errorHandler.
RxJavaPlugins.setErrorHandler(new Consumer<Throwable>() {
@Override
public void accept(Throwable throwable) throws Exception {
LogUtil.e("RxJava catch global exception", throwable);
}
});
这样就可以保障RxJava2自己的调用链流程不会崩溃.
但是如果我们希望有了异常,还能进入自己的onError做对应处理.那么也很简单.对自己传入的observer做一次封装,直接try住自己的onNext块,然后发给onError就行.
说到这里难免就要提到rxjava1.rxjava1的没有崩溃正式因为如此.rxjava1中subscribe一个subscriber,他并没有直接使用这个subscriber,而是直接对他封装成SafeSubscriber,可以看到,就是一个简单地try操作.然后转onError
@Override
public void onNext(T t) {
try {
if (!done) {
actual.onNext(t);
}
} catch (Throwable e) {
// we handle here instead of another method so we don't add stacks to the frame
// which can prevent it from being able to handle StackOverflow
Exceptions.throwOrReport(e, this);
}
}
public static void throwOrReport(Throwable t, Observer<?> o) {
Exceptions.throwIfFatal(t);
o.onError(t);
}
说道这里,其实真个流程也算是又摸索了一遍,算是又熟悉一遍RxJava的链式流程.不过这其实这是最微小的一部分.Rxjava操作符用这么点,实在是不应该.
另外就是说这么多,异常的根本原因,还是自身业务代码的糟糕.这一点也无力吐槽了.正如使用这个RxJava一样,根本没有展示RxJava最强大的一面.只是把他做成了了最普通的回调功能.业务层不使用RxJava来消除无限回调,那么使用Rxjava的意义也缩小了.另外就是,代码的规范看上去远比使用新得技术来的重要.
盗图一张(其实我也想画一张,但是这幅感觉已经足够完美,这也是我要偷过来的原因):
这幅图比我上面打那么多字给力的多.但我保证,我摸这个流程的时候,没过看这个图.
Observable.create().map().doOnNext().subsribeOn().observeOn().subscibe(Observe);
如上所述这个链式调用:
RxJava任务调用链
最后说三句话来总结,我的总结完了.谢谢.
