迷之RxJava (三)—— 线程切换
来自: http://segmentfault.com/a/1190000004051191
Rxjava -- 一个异步库
RxJava
最迷人的是什么?
答案就是把异步序列写到一个工作流里!
和javascript
的Promise/A
如出一辙。
OK,在java
中做异步的事情在我们传统理解过来可不方便,而且,如果要让异步按照我们的工作流来,就更困难了。
但是在RxJava
中,我们只要调用调用
subscribOn()
和observeOn()
就能切换我们的工作线程,是不是让小伙伴都惊呆了?
然后结合RxJava
的Operator
,写异步的时候,想切换线程就是一行代码的事情,整个workflow
还非常清晰:
Observable.create() // do something on io thread .work() // work.. work.. .subscribeOn(Schedulers.io()) // observeOn android main thread .observeOn(AndroidSchedulers.mainThread()) .subscribe();
我们再也不用去写什么见鬼的new Thread
和Handler
了,在这么几行代码里,我们实现了在io
线程上做我们的工作(work
),在main
线程上,更新UI
Subscribe On
先看下subscribeOn
干了什么
public final Observable<T> subscribeOn(Scheduler scheduler) { if (this instanceof ScalarSynchronousObservable) { return ((ScalarSynchronousObservable<T>)this).scalarScheduleOn(scheduler); } return nest().lift(new OperatorSubscribeOn<T>(scheduler)); }
啊,原来也是个lift,就是从一个Observable
生成另外一个Observable
咯,这个nest
是干嘛用?
public final Observable<Observable<T>> nest() { return just(this); }
这里返回类型告诉我们,它是产生一个Observable<Observable<T>>
讲到这里,会有点晕,先记着这个,然后我们看OperatorSubscribeOn
这个操作符,
构造函数是
public OperatorSubscribeOn(Scheduler scheduler) { this.scheduler = scheduler; }
OK,这里保存了scheduler
对象,然后就是我们前一章说过的转换方法。
@Override public Subscriber<? super Observable<T>> call(final Subscriber<? super T> subscriber) { final Worker inner = scheduler.createWorker(); subscriber.add(inner); return new Subscriber<Observable<T>>(subscriber) { @Override public void onCompleted() { // ignore because this is a nested Observable and we expect only 1 Observable<T> emitted to onNext } @Override public void onError(Throwable e) { subscriber.onError(e); } @Override public void onNext(final Observable<T> o) { inner.schedule(new Action0() { @Override public void call() { final Thread t = Thread.currentThread(); o.unsafeSubscribe(new Subscriber<T>(subscriber) { @Override public void onCompleted() { subscriber.onCompleted(); } @Override public void onError(Throwable e) { subscriber.onError(e); } @Override public void onNext(T t) { subscriber.onNext(t); } @Override public void setProducer(final Producer producer) { subscriber.setProducer(new Producer() { @Override public void request(final long n) { if (Thread.currentThread() == t) { // don't schedule if we're already on the thread (primarily for first setProducer call) // see unit test 'testSetProducerSynchronousRequest' for more context on this producer.request(n); } else { inner.schedule(new Action0() { @Override public void call() { producer.request(n); } }); } } }); } }); } }); } }; }
让人纠结的类模板
看完这段又臭又长的,先深呼吸一口气,我们慢慢分析下。
首先要注意RxJava
里面最让人头疼的模板问题,那么OperatorMap
这个类的声明是
public final class OperatorMap<T, R> implements Operator<R, T>
而Operator
这个接口继承Func1
public interface Func1<T, R> extends Function { R call(T t); }
我们这里不要记T
和R
,记住传入左边的模板是形参,传入右边的模板是返回值
。
好了,那么这里的
call
就是从一个T
转换成一个Observable<T>
的过程了。
总结一下,我们这一次调用subscribeOn
,做了两件事
1、
nest()
为Observable<T>
生成了一个Observable<Observable<T>>
2、lift()
对Observalbe<Observalbe<T>>
进行一个变化,变回Observable<T>
因为lift
是一个模板函数,它的返回值的类型是参照它的形参来,而他的形参是Operator<T, Observable<T>>
这个结论非常重要!!
OK,到这里我们已经存储了所有的序列,等着我们调用了。
调用链
首先,记录我们在调用这条指令之前的Observable<T>
,记为Observable$1
然后,经过lift
生成的Observable<T>
记为Observable$2
好了,现在我们拿到的依然是Observable<T>
这个对象,但是它不是原始的Observable$1
,要深深记住这一点,它是由lift
生成的Observable$2
,这时候进行subscribe
,那看到首先调用的就是OnSubscribe.call
方法,好,直接进入lift
当中生成的那个地方。
我们知道这一层lift
的operator
就是刚刚的OperatorSubscribOn
,那么调用它的call
方法,生成的是一个Subscriber<Observable<T>>
Subscriber<? super T> st = hook.onLift(operator).call(o); try { // new Subscriber created and being subscribed with so 'onStart' it st.onStart(); onSubscribe.call(st); } catch (Throwable e) { ... }
好,还记得我们调用过nest
么?,这里的onSubscribe
可是nest
上下文中的噢,每一次,到这个地方,这个onSubscribe
就是上一层Observable
的onSubscribe
,即Observable<Observable<T>>
的onSubscribe
,相当于栈弹出了一层。它的call
直接在Subscriber
的onNext
中给出了最开始的Observable<T>
,我们这里就要看下刚刚在OperatorSubscribeOn
中生成的Subscriber
new Subscriber<Observable<T>>(subscriber) { @Override public void onCompleted() { // ignore because this is a nested Observable and we expect only 1 Observable<T> emitted to onNext } @Override public void onError(Throwable e) { subscriber.onError(e); } @Override public void onNext(final Observable<T> o) { inner.schedule(new Action0() { @Override public void call() { final Thread t = Thread.currentThread(); o.unsafeSubscribe(new Subscriber<T>(subscriber) { @Override public void onCompleted() { subscriber.onCompleted(); } @Override public void onError(Throwable e) { subscriber.onError(e); } @Override public void onNext(T t) { subscriber.onNext(t); } }); } }); } }
对,就是它,这里要注意,这里的subscriber
就是我们在lift
中,传入的o
Subscriber<? super T> st = hook.onLift(operator).call(o);
对,就是它,其实它就是SafeSubscriber
。
回过头,看看刚刚的onNext()
方法,inner.schedule()
这个函数,我们可以认为就是postRun()
类似的方法,而onNext()
中传入的o
是我们之前生成的Observable$1
,是从Observable.just
封装出来的Observable<Observable<T>>
中产生的,这里调用了Observable$1.unsafeSubscribe
方法,我们暂时不关心它和subscribe
有什么不同,但是我们知道最终功能是一样的就好了。
注意它运行时的线程!!在
inner
这个Worker
上!于是它的运行线程已经被改了!!
好,这里的unsafeSubscribe
调用的方法就是调用原先Observable$1.onSubscribe
中的call
方法:
这个Observable$1
就是我们之前自己定义的Observable
了。
综上所述,如果我们需要我们的Observable$1
在一个别的线程上运行的时候,只需要在后面跟一个subscribeOn
即可。结合扔物线大大的图如下:
总结
这里逻辑着实不好理解。如果还没有理解的朋友,可以按照我前文说的顺序,细致的看下来,我把逻辑过一遍之后,发现lift
的陷阱实在太大,内部类用的风生水起,一不小心,就不知道一个变量的上下文是什么,需要特别小心。
迷之RxJava(四)—— Retrofit和RxJava的基情
本文在不停更新中,如果有不明白的地方(可能会有很多),请大家给出意见,拍砖请轻点= =