?首先我提出來兩個問題：
1.Rxjava是如何做到執行緒切換的
2.執行緒切換我多次呼叫subscribeOn和多次呼叫observeOn，對資料流有什么影響
對Rxjava的資料處理流程不是很清楚的，可以看我上一篇文章
資料處理及訂閱流程分析
下面這段代碼相信很多人都寫過
upstream.subscribeOn(SchedulerProvider.net())
.observeOn(AndroidSchedulers.mainThread());
上游的Observable進行執行緒切換在子執行緒發射，然后訂閱者代碼運行到主執行緒，
接下來我們看一下subscribeOn的原始碼
public final Observable<T> subscribeOn(Scheduler scheduler) {
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this, scheduler));
}
創建了一個ObservableSubscribeOn并回傳，把自身與scheduler通過構造方法傳入到ObservableSubscribeOn，顯然ObservableSubscribeOn一定繼承了Observable.這里我們發現，向下游傳遞的Observable已經被替換成了ObservableSubscribeOn.上游的Observable被ObservableSubscribeOn接管了，又是代理模式的應用，分析ObservableSubscribeOn
public final class ObservableSubscribeOn<T> extends AbstractObservableWithUpstream<T, T> {
final Scheduler scheduler;
public ObservableSubscribeOn(ObservableSource<T> source, Scheduler scheduler) {
super(source);
this.scheduler = scheduler;
}
@Override
public void subscribeActual(final Observer<? super T> observer) {
final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(observer);
observer.onSubscribe(parent);
parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
}
}
構造方法里把上游的observable賦值到自己的source成員變數，同時把Scheduler保存到自己的成員變數，然后唯一的功能方法subscribeActual（observer）,非常得簡潔，把下游的observer通過SubscribeOnObserver代理，然后呼叫parent的setDisposable方法，引數為scheduler.scheduleDirect(new SubscribeTask(parent))，
我們一看SubscribeOnObserver這個observer的原始碼便能清楚是什么鬼了
static final class SubscribeOnObserver<T> extends AtomicReference<Disposable> implements Observer<T>, Disposable {
private static final long serialVersionUID = 8094547886072529208L;
final Observer<? super T> downstream;
final AtomicReference<Disposable> upstream;
SubscribeOnObserver(Observer<? super T> downstream) {
this.downstream = downstream;
this.upstream = new AtomicReference<Disposable>();
}
@Override
public void onSubscribe(Disposable d) {
DisposableHelper.setOnce(this.upstream, d);
}
@Override
public void onNext(T t) {
downstream.onNext(t);
}
@Override
public void onError(Throwable t) {
downstream.onError(t);
}
@Override
public void onComplete() {
downstream.onComplete();
}
@Override
public void dispose() {
DisposableHelper.dispose(upstream);
DisposableHelper.dispose(this);
}
@Override
public boolean isDisposed() {
return DisposableHelper.isDisposed(get());
}
void setDisposable(Disposable d) {
DisposableHelper.setOnce(this, d);
}
}
沒幾行代碼，也很簡潔，關注一下setDisposable（Disposable ）方法，其實就是把scheduler回傳的Disposable賦值給自己，也就是說ObservableSubscribeOn可以dispose,scheduler回傳的那個disposable.通過我的上篇文章資料處理及訂閱流程分析我們知道ObservableSubscribeOn的subscribeActual方法執行發生在資料源被訂閱的時候，也就是說訂閱的時候把下游的observer包裝為一個SubscribeTask，在scheduler的scheduleDirect方法中執行，
final class SubscribeTask implements Runnable {
private final SubscribeOnObserver<T> parent;
SubscribeTask(SubscribeOnObserver<T> parent) {
this.parent = parent;
}
@Override
public void run() {
source.subscribe(parent);
}
}
?這里的source就是上游的Observable，上游的observable的subscribe方法在run方法中運行，scheduler作為后續文章講，它的scheduleDirect方法就是把SubscribeTask運行在子執行緒，
?再梳理一遍：當最下游的observable發生訂閱的時候 -->會呼叫ObservableSubscribeOn的subscribeActual（observer）方法，這里的引數就是我們下游的observer，這個observer被包裝后作為一個task，上游的source.subscribe（parent）方法運行在子執行緒，parent實體為SubscribeOnObserver,，上游source.subscribe方法內部實作，都會想方設法執行parent的onNext,onComplete等方法，就達成了包裝類SubscribeOnObserver（即parent）這些方法在子執行緒運行，同時回傳一個disposable,這個disposable給Observer用來取消訂閱，又有些饒了，其實一個時序圖就能說明一切，點擊查看大圖
?建議讀者跟著原始碼一起看，思考一下，就明白了，
?回答第二個問題，多次到用subscribeOn會怎樣？subscribeOn會把下游的Observer方法的onNext，onComplete，onError方法包裝在SubscribeOnObserver方法中，這些方法會在訂閱時運行在子執行緒的task中，多次呼叫會多次包裝下游的Observer,進行執行緒切換到子執行緒，那多次切換最侄訓切換成啥樣呢？其實仔細想一下，答案就在這段代碼中
@Override
public void subscribeActual(final Observer<? super T> observer) {
final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(observer);
observer.onSubscribe(parent);
parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
}
下游的subscrbeActual先執行，當然最靠近上游的，是程式最后一次執行subscrbeActual方法啦，就是第一次呼叫subscribeOn在哪個執行緒中執行，最終結果會在哪個執行緒中執行，這個要牢記下游的subscribeActual最先執行就能知道結果了，因為每次subscribeActual都會切換執行緒（僅針對ObservableSubscribeOn里的方法），
?接下來分析observerOn方法
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回傳，老樣子創建了一個新的Observable向下傳遞，看ObservableSubscribeOn原始碼
public final class ObservableObserveOn<T> extends AbstractObservableWithUpstream<T, T> {
final Scheduler scheduler;
final boolean delayError;
final int bufferSize;
public ObservableObserveOn(ObservableSource<T> source, Scheduler scheduler, boolean delayError, int bufferSize) {
super(source);
this.scheduler = scheduler;
this.delayError = delayError;
this.bufferSize = bufferSize;
}
@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));
}
}
}
構造方法與上面分析相似 ，直接看subscribeActual方法，暫時不關心scheduler instanceof TrampolineScheduler的情況，接下來就是上游Observable source訂閱ObserveOnObserver，看ObserveOnObserver
static final class ObserveOnObserver<T> extends BasicIntQueueDisposable<T>
implements Observer<T>, Runnable {
private static final long serialVersionUID = 6576896619930983584L;
final Observer<? super T> downstream;
final Scheduler.Worker worker;
final boolean delayError;
final int bufferSize;
SimpleQueue<T> queue;
Disposable upstream;
Throwable error;
volatile boolean done;
volatile boolean disposed;
int sourceMode;
boolean outputFused;
ObserveOnObserver(Observer<? super T> actual, Scheduler.Worker worker, boolean delayError, int bufferSize) {
this.downstream = actual;
this.worker = worker;
this.delayError = delayError;
this.bufferSize = bufferSize;
}
@Override
public void onSubscribe(Disposable d) {
if (DisposableHelper.validate(this.upstream, d)) {
this.upstream = d;
if (d instanceof QueueDisposable) {
@SuppressWarnings("unchecked")
QueueDisposable<T> qd = (QueueDisposable<T>) d;
int m = qd.requestFusion(QueueDisposable.ANY | QueueDisposable.BOUNDARY);
if (m == QueueDisposable.SYNC) {
sourceMode = m;
queue = qd;
done = true;
downstream.onSubscribe(this);
schedule();
return;
}
if (m == QueueDisposable.ASYNC) {
sourceMode = m;
queue = qd;
downstream.onSubscribe(this);
return;
}
}
queue = new SpscLinkedArrayQueue<T>(bufferSize);
downstream.onSubscribe(this);
}
}
@Override
public void onNext(T t) {
if (done) {
return;
}
if (sourceMode != QueueDisposable.ASYNC) {
queue.offer(t);
}
schedule();
}
@Override
public void onError(Throwable t) {
if (done) {
RxJavaPlugins.onError(t);
return;
}
error = t;
done = true;
schedule();
}
@Override
public void onComplete() {
if (done) {
return;
}
done = true;
schedule();
}
@Override
public void dispose() {
if (!disposed) {
disposed = true;
upstream.dispose();
worker.dispose();
if (getAndIncrement() == 0) {
queue.clear();
}
}
}
@Override
public boolean isDisposed() {
return disposed;
}
void schedule() {
if (getAndIncrement() == 0) {
worker.schedule(this);
}
}
....省略一些
}
代碼稍微有些多，看schedule()方法，worker.schedule(this);拿worker執行自己的runable方法，runable方法
@Override
public void run() {
if (outputFused) {
drainFused();
} else {
drainFused();
}
}
有興趣的可以分析drainFused，drainFused，我在以后的文章中會單獨分析這drain相關方法，其實它就是呼叫了下游對應的onNext,onCompete,onError等方法，
梳理一下：也就是說我們每次呼叫observeOn(scheduler)方法，都會包裝一個新的Observable向下傳遞，同時把下游的Observer包裝成ObserveOnObserver，又是代理，通過它里面的worker在上游的Observer執行onNext等方法，在ObserveOnObserver的onNext做執行緒切換，呼叫下游的observer的onNext方法，從而實作執行緒切換，
那么多次呼叫observeOn方法時，會怎樣呢？思考一下，其實很容易了，Observer的發射方法呼叫順序是從上游，一直傳遞到包裝下游，也就是說最后一次執行執行緒切換有效，當然中間的執行緒切換也有效果，只是最后一次Observer為我們傳遞的Observer而已，所以我們的Observer方法的onNext相關方法在哪個執行緒，就取決于最后一次呼叫了，
備注：Observer的onSubscribe方法，是從上游向下游呼叫的，這一點恰好跟subscribeOnActual相反，仔細想一下就知道了，其實說一千道一萬，也沒有自己梳理一遍強，
轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/172437.html
標籤：其他