这篇文章的主要目的是让你能够通过图形以及时序图，清晰的了解Rxjava2的调用流程以及数据流向，下面先贴段我们要分析的代码：
//1.如何创建一个上游
Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
e.onNext(1);
}
});
Observer<Integer> observer = new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.e(TAG, "onSubscribe: d==" + Thread.currentThread().getName());
}
@Override
public void onNext(Integer integer) {
Log.e(TAG, "onNext: o==" + integer + " " + Thread.currentThread().getName());
}
@Override
public void onError(Throwable e) {
Log.e(TAG, "error");
}
@Override
public void onComplete() {
Log.e(TAG, "onComplete:");
}
};
observable.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(observer);
上面的代码的意图很简单，我们在IO线程中发送一个Integer类型的整数 ，然后在UI线程中将其打印出来，这里面涉及到了线程的切换，在分析之前我们先贴张图和一张时序图 ，来帮助我们理解分析流程
该图片来自网络上一位博主:kJ的专栏 然后我自己添加了些备注
接下来贴一下我自己画的时序图 ，然后我们按照时序图一步一步来分析
我们首先从头开始分析：
1.我们需要首先调用Observable的create方法来创建一个上游的数据流对象
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
ObjectHelper.requireNonNull(source, "source is null");
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}```
2.create方法调用了ObservableOnSubscribe的构造方法创建了一个匿名内部类对象并且重写了该对象的subscribe方法
3.然后以该对象为参数创建了ObservableCreate对象 并且将source保存到了自己的变量source中 这样ObservableCreate引用了对象ObservableOnSubscribe对象，方便后边调用该对象的方法
4.上面的create方法返回了一个ObservableCreate对象，接下来我们调用这个对象的
subscribeOn方法 并且将Schedulers.io()返回的对象作为该方法的参数，
但是我们发现 ObservableCreate对象中并没有subscribeOn方法 那肯定是父类Observable的方法 我们来看下
public final Observable<T> subscribeOn(Scheduler scheduler) {
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this, scheduler));
}
这里又以上面创建的ObservableCreate和用来切换线程使用的Scheduler对象为参数创建了一个ObservableSubscribeOn对象并且返回
5.我们获得了第二步创建的ObservableSubscribeOn对象后接着调用observeOn(AndroidSchedulers.mainThread()) 方法 来切换回主线程 ，我们发现ObservableSubscribeOn对象也没有observeOn方法 此方法也是父类Observable的方法
public final Observable<T> observeOn(Scheduler scheduler) {
return observeOn(scheduler, false, bufferSize());
}
调用了重载的方法
public final Observable<T> observeOn(Scheduler scheduler, boolean delayError, int bufferSize) {
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
ObjectHelper.verifyPositive(bufferSize, "bufferSize");
return RxJavaPlugins.onAssembly(new ObservableObserveOn<T>(this, scheduler, delayError, bufferSize));
}```
我们发现又创建了一个新的对象ObservableObserveOn 并且保存了上面创建的
ObservableSubscribeOn对象以及UI线程的Scheduler 并且返回
6.最后将我们的Observable和observer关联起来 ，调用了ObservableObserveOn 对象的subscribe方法 并且 以Observer为参数 而这个subscribe方法是父类Observable的方法 所以最终调用到了这里
public final void subscribe(Observer<? super T> observer) {
.....
observer = RxJavaPlugins.onSubscribe(this, observer);//这个不用关心 就当成传进去什么东西 就返回什么东西 所以Observer还是参数Observer并没有什么改变
ObjectHelper.requireNonNull(observer, "Plugin returned null Observer");
subscribeActual(observer);//重点是在这里 这是个接口方法 所以会去找它的实现 但是实现是谁呢？
.....
我们先对上面的步骤 做一个总结：
1.我们创建了一个ObservableOnSubscribe匿名内部类对象 我们简化为OOS对象
2.然后将 OOS对象保存到了新创建的ObservableCreate对象的source变量中
3.调用subscribeOn方法 又 创建了Schedduler.IO对象（用来切换到IO线程的 这个IO线程由线程池控制）和ObservableSubscribeOn 对象
将上面创建的ObservableCreate对象OC和IO保存到了ObservableSubscribeOn对象的
public ObservableSubscribeOn(ObservableSource<T> source, Scheduler scheduler) {
super(source);//this.source=source 对象类型为ObservableCreate
this.scheduler = scheduler;
}
4.接着调用Observable的observerOn方法 切换回UI线程 在这个方法里面创建了ObservableObserveOn对象
public ObservableObserveOn(ObservableSource<T> source, Scheduler scheduler, boolean delayError, int bufferSize) {
super(source);//source类型为上面创建的ObservableSubscribeOn
this.scheduler = scheduler;//实际类型是时序图中的HandlerSchedudler对象 具体源码自行分析
this.delayError = delayError;//暂不关注
this.bufferSize = bufferSize;
}
5.然后最后调用subcribe方法建立订阅关系
我们接着往下分析 ：
7.上面第五步 我们调用了subcribe方法 这里的this指向的是我们第四步创建的ObservableObserveOn对象 所以最后调用的是该对象对subscribeActual方法的实现
我们看下实现代码：
@Override
protected void subscribeActual(Observer<? super T> observer) {
if (scheduler instanceof TrampolineScheduler) {//这里If条件不成立
source.subscribe(observer);
} else {
//所以会走下面的逻辑
Scheduler.Worker w = scheduler.createWorker();//调用HandlerSchedudler的crateWorker方法 返回一个HandlerWorker对象
source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
// 然后调用source的subscribe方法 我们这里需要关注两点：
1.source的实际类型是什么//这里回头看下总结的第4步-- 实际类型是ObservableSubscribeOn
2.参数ObserveOnOberver对Observer进行了封装
}
}
9.这里我们进入ObservableSubscribeOn的subscribe方法 （该方法实际上是Observable的方法 最后会调用到子类的subscribeActual方法）我们直接看subscribeActual方法
public void subscribeActual(final Observer<? super T> s) {
//这里传进来的S类型为ObserveOnOberver 并且里面保存了最初代码中创建的oberver以及线程切换对象HandlerWorker
//这里又对ObserveOnOberver对象进行一次封装
final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(s);
s.onSubscribe(parent);
// 这里的scheduler对象是Schedulers.IO创建的 所以此时会切换到该IO线程中去执行source.subscribe(parent);
parent.setDisposable(scheduler.scheduleDirect(new Runnable() {
@Override
public void run() {
source.subscribe(parent);//最后会执行到这里
1.source对象指向了ObservableCreate对象 如果忘了回到第二步看下
那么最终会调用ObservableCreate的subscribeActua方法 注意这里的参数是SubscribeOnObserver
}
}));
}```
10.这一步调用到了ObservableCreate的subscribeActual
@Override
protected void subscribeActual(Observer<? super T> observer) {
//observer对象类型SubscribeOnObserver
//创建了CreateEmitter对象 并且传递给subscribe方法
CreateEmitter<T> parent = new CreateEmitter<T>(observer);
observer.onSubscribe(parent);
try {
source.subscribe(parent);//这里source指向了最初创建的匿名内部类ObservableOnSubscribe
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
}
这里我们回头看下create的方法
Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
e.onNext(1);//最后走到了这里 调用传进来的e的onNext方法发送数据
//这里e指向CreateEmitter
}
});
好了我们终于走到了发送数据的地方了 。
这调用方向实际上是由下游往上游逐步调用的 接下来就是从上游到下游逐步再回去了，我们接下来就分析 如何将发送的数据显示到UI上的。
这里我们看下CreateEmitter的onNext方法
12.
@Override
public void onNext(T t) {
if (t == null) {
onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
return;
}
if (!isDisposed()) {
observer.onNext(t);//这里主要关注Observer的实际类型是什么？我们在第10步创建了一个SubscribeOnObserver对象并且保存到了CreateEmitter的变量Observer变量中
}
}```
13.我们继续往下分析SubscribeOnObserver的onNext方法
@Override
public void onNext(T t) {
actual.onNext(t);//这个actual对象实际指向ObserveOnObserver 这个对象是在调用ObservableObserveOn的subscribeActual方法里面创建的
}
14.接下来就跳到ObserveOnObserver 的onNext方法中执行方法
@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);//这里的worker实际上就是我们前边创建的HandlerWorker对象了 该对象关联了一个持有UI线程Looper的Handler 可以往主线程中发送消息
}
}
@Override
public Disposable schedule(Runnable run, long delay, TimeUnit unit) {
.....
ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);
Message message = Message.obtain(handler, scheduled);
message.obj = this; // Used as token for batch disposal of this worker's runnables.
handler.sendMessageDelayed(message, Math.max(0L, unit.toMillis(delay)));
//我们看到往主线程发送了一个Message对象 然后会执行run的run方法
参数run指向发起调用的ObserveOnObserver 对象 所以会调用该对象的run方法
// Re-check disposed state for removing in case we were racing a call to dispose().
if (disposed) {
handler.removeCallbacks(scheduled);
return Disposables.disposed();
}
return scheduled;
}
@Override
public void run() {
if (outputFused) {
drainFused();
} else {
drainNormal();//这里会走这里的逻辑
}
}
接下来我们就来分析drainNormal方法
void drainNormal() {
int missed = 1;
final SimpleQueue<T> q = queue;
final Observer<? super T> a = actual;
for (;;) {
if (checkTerminated(done, q.isEmpty(), a)) {
return;
}
for (;;) {
boolean d = done;
T v;
try {
v = q.poll();
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
s.dispose();
q.clear();
a.onError(ex);
worker.dispose();
return;
}
boolean empty = v == null;
if (checkTerminated(d, empty, a)) {
return;
}
if (empty) {
break;
}
a.onNext(v);//关键在这里 调用a的onNext方法此时我们应该感觉到快到底了 坚持一下
}
missed = addAndGet(-missed);
if (missed == 0) {
break;
}
}
}```
上面代码中的a对象实际上指向了哪个对象呢 actual这个对象我们看看是该类的成员变量
ObserveOnObserver(Observer<? super T> actual, Scheduler.Worker worker, boolean delayError, int bufferSize) {
this.actual = actual;//在这里赋值我们也知道ObserveOnObserver的构造函数是在ObservableObserveOn方法里调用的 传进来的就是我们自己定义的Observer对象 就是最初的代码中创建的Observer对象
......
}```
我们这边再回过头看下；
Observer<Integer> observer = new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.e(TAG, "onSubscribe: d==" + Thread.currentThread().getName());
}
@Override
public void onNext(Integer integer) {
Log.e(TAG, "onNext: o==" + integer + " " + Thread.currentThread().getName());
}
@Override
public void onError(Throwable e) {
Log.e(TAG, "error");
}
@Override
public void onComplete() {
Log.e(TAG, "onComplete:");
}
};```
最后会调用该对象的onNext方法 最后将结果打印出来
好了 ，我们整个的流程分析完了 ，希望能帮到大家 谢谢！