介绍
observeOn(线程B)方法是将数据处理后的结果放置在线程B内执行。通常线程B是UI线程
执行代码
//初始化被观察者Observable，并给其加上数据处理器Observable.OnSubscribe
Observable Aobservable = Observable.create(new Observable.OnSubscribe<String>(){
@Override
public void call(Subscriber<? super String> subscriber) {
LogShowUtil.addLog("RxJava","发送线程: "+Thread.currentThread().getName(),true);
subscriber.onNext("杨");
subscriber.onCompleted();
}
});
//做subscribeOn数据处理器执行线程切换处理
Observable Bobservable = Aobservable.subscribeOn(Schedulers.newThread());
//做observeOn结果接受线程切换处理
Observable Cobservable = Bobservable.observeOn(AndroidSchedulers.mainThread());
//订阅
Cobservable.subscribe(observer);
源码分析
1. 初始化被观察者AObservable
Observable Aobservable = Observable.create(原始数据处理器);
由此可知被观察者AObservable持有原始数据处理器对象Observable.OnSubscribe。
2. 执行subscribeOn数据处理线程切换操作
Observable Bobservable = Aobservable.subscribeOn(Schedulers.newThread());
Observable#subscribeOn
public final Observable<T> subscribeOn(Scheduler scheduler) {
if (this instanceof ScalarSynchronousObservable) {
return ((ScalarSynchronousObservable<T>)this).scalarScheduleOn(scheduler);
}
return create(new OperatorSubscribeOn<T>(this, scheduler));
}
接着我们看其中的new OperatorSubscribeOn(Aobservable,线程切换工具)操作
public OperatorSubscribeOn(Observable<T> source, Scheduler scheduler) {
this.scheduler = scheduler;
this.source = source;
}
由代码可知代理数据处理线程切换器OperatorSubscribeOn持有
Aobservable和线程切换工具Scheduler
回到subscribeOn()方法内继续执行create(代理据处理线程切换器)
return create(new OperatorSubscribeOn<T>(this, scheduler));
create方法之前已经分析过，由此可知Bobservable持有代理据处理线程切换器OperatorSubscribeOn。
3. 执行subscribeOn结果接收线程切换操作
Observable Cobservable = Bobservable.observeOn(AndroidSchedulers.mainThread());
public final Observable<T> observeOn(Scheduler scheduler) {
return observeOn(scheduler, RxRingBuffer.SIZE);
}
public final Observable<T> observeOn(Scheduler scheduler, int bufferSize) {
return observeOn(scheduler, false, bufferSize);
}
public final Observable<T> observeOn(Scheduler scheduler, boolean delayError, int bufferSize) {
if (this instanceof ScalarSynchronousObservable) {
return ((ScalarSynchronousObservable<T>)this).scalarScheduleOn(scheduler);
}
return lift(new OperatorObserveOn<T>(scheduler, delayError, bufferSize));
}
接着我们看其中的new OperatorObserveOn<T>(结果接受线程切换工具, delayError, bufferSize)操作
public OperatorObserveOn(Scheduler scheduler, boolean delayError, int bufferSize) {
this.scheduler = scheduler;
this.delayError = delayError;
this.bufferSize = (bufferSize > 0) ? bufferSize : RxRingBuffer.SIZE;
}
由代码可知代理结果接受线程切换器OperatorObserveOn持有结果接受线程切换工具Scheduler
接着回到observeOn()方法内的lift()方法
public final Observable<T> observeOn(Scheduler scheduler, boolean delayError, int bufferSize) {
if (this instanceof ScalarSynchronousObservable) {
return ((ScalarSynchronousObservable<T>)this).scalarScheduleOn(scheduler);
}
return lift(代理结果接受线程切换器OperatorObserveOn);
}
public final <R> Observable<R> lift(final Operator<? extends R, ? super T> operator) {
return create(new OnSubscribeLift<T, R>(onSubscribe, operator));
}
接着我们看new OnSubscribeLift()方法
public OnSubscribeLift(OnSubscribe<T> parent, Operator<? extends R, ? super T> operator) {
this.parent = parent;
this.operator = operator;
}
由此可知Cobservable持有Bobservable和代理结果接受线程切换器OperatorObserveOn。
4. 初始化结果接受器观察者Observer
Cobservable.subscribe(observer);
由之前分析可知会使用 Cobservable内的OnSubscribeLift做call()方法。
其中observer为结果接受器。
@Override
public void call(Subscriber<? super R> o) {
try {
//步骤一 获取结果接受线程切换器
Subscriber<? super T> st = RxJavaHooks.onObservableLift(operator).call(o);
try {
st.onStart();
//步骤二 执行Bobservable.call()方法。
parent.call(st);
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
st.onError(e);
}
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
o.onError(e);
}
}
首先看步骤一，其实就是执行代理结果接受线程切换器OperatorObserveOn的call()方法，其中o参数为结果接收器Observer
OperatorObserveOn#call
@Override
public Subscriber<? super T> call(Subscriber<? super T> child) {
if (scheduler instanceof ImmediateScheduler) {
// avoid overhead, execute directly
return child;
} else if (scheduler instanceof TrampolineScheduler) {
// avoid overhead, execute directly
return child;
} else {
ObserveOnSubscriber<T> parent = new ObserveOnSubscriber<T>(scheduler, child, delayError, bufferSize);
parent.init();
return parent;
}
}
经过判断会进入new ObserveOnSubscriber()方法生成结果接受线程切换器
ObserveOnSubscriber，其中主要参数为结果接受线程切换工具scheduler和结果接收器child。
public ObserveOnSubscriber(Scheduler scheduler, Subscriber<? super T> child, boolean delayError, int bufferSize) {
this.child = child;
this.recursiveScheduler = scheduler.createWorker();
this.delayError = delayError;
this.on = NotificationLite.instance();
int calculatedSize = (bufferSize > 0) ? bufferSize : RxRingBuffer.SIZE;
// this formula calculates the 75% of the bufferSize, rounded up to the next integer
this.limit = calculatedSize - (calculatedSize >> 2);
if (UnsafeAccess.isUnsafeAvailable()) {
queue = new SpscArrayQueue<Object>(calculatedSize);
} else {
queue = new SpscAtomicArrayQueue<Object>(calculatedSize);
}
// signal that this is an async operator capable of receiving this many
request(calculatedSize);
}
由上面代码分析可知结果接受线程切换器
ObserveOnSubscriber持有结果接受线程切换工具scheduler和结果接收器Observer。
再回到OnSubscribeLift#call方法内的步骤二操作
@Override
public void call(Subscriber<? super R> o) {
try {
//步骤一 获取结果接受线程切换器
Subscriber<? super T> st = RxJavaHooks.onObservableLift(operator).call(o);
try {
st.onStart();
//步骤二 执行Bobservable.call()方法。
parent.call(st);
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
st.onError(e);
}
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
o.onError(e);
}
}
步骤二将执行Bobservable.call(st)方法，其中参数
st就是我们刚刚生成的接受线程切换器ObserveOnSubscriber。
执行步骤二做的操作通过上一篇辅助操作符subscribeOn()可以知道会开启一个新的线程去做数据处理操作，并把Bobservable的接收器替换为了结果接受线程切换器ObserveOnSubscriber。最终处理完数据会会调用ObserveOnSubscriber.onNext()方法返回数据。
ObserveOnSubscriber#onNext
@Override
public void onNext(final T t) {
if (isUnsubscribed() || finished) {
return;
}
//将结果存起来
if (!queue.offer(on.next(t))) {
onError(new MissingBackpressureException());
return;
}
//执行结果线程切换
schedule();
}
由上面代码可知，会先数据处理结果存储起来，然后执行
schedule()方法
protected void schedule() {
if (counter.getAndIncrement() == 0) {
recursiveScheduler.schedule(this);
}
}
上面代码执行完以后会回调进入ObserveOnSubscriber#call()方法
此时call()方法就通过结果接受线程切换工具切换至Bobservable.observeOn(线程B)所设置的线程B内了
@Override
public void call() {
long missed = 1L;
long currentEmission = emitted;
final Queue<Object> q = this.queue;
final Subscriber<? super T> localChild = this.child;
final NotificationLite<T> localOn = this.on;
for (;;) {
long requestAmount = requested.get();
while (requestAmount != currentEmission) {
boolean done = finished;
Object v = q.poll();
boolean empty = v == null;
if (checkTerminated(done, empty, localChild, q)) {
return;
}
if (empty) {
break;
}
//结果接收器执行onNext()方法
localChild.onNext(localOn.getValue(v));
currentEmission++;
if (currentEmission == limit) {
requestAmount = BackpressureUtils.produced(requested, currentEmission);
request(currentEmission);
currentEmission = 0L;
}
}
if (requestAmount == currentEmission) {
if (checkTerminated(finished, q.isEmpty(), localChild, q)) {
return;
}
}
emitted = currentEmission;
missed = counter.addAndGet(-missed);
if (missed == 0L) {
break;
}
}
}
通过上面代码可知，call()方法内通过自己已经持有的结果接收器
Observer执行localChild.onNext(localOn.getValue(v))操作
接着会进入结果接收器Observer内方法体内
Observer observer = new Observer<String>() {
@Override
public void onCompleted() {
LogShowUtil.addLog("RxJava","结束",true);
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(String string) {
LogShowUtil.addLog("RxJava","接受线程: "+Thread.currentThread().getName(),true);
LogShowUtil.addLog("RxJava","结果: "+string,true);
}
};
最终输出结果
发送线程: RxNewThreadScheduler-1
接受线程: main
结果: 杨
结束
总结
通过上面代码跟踪可知结果接受线程切换的具体位置是在OperatorObserveOn#ObserveOnSubscriber#onNext()的方法内执行完schedule()后。
网友评论