Android之Rxjava2-源碼解析，flutter安裝包優化

public static boolean isDisposed(Disposable d) {
//判斷Disposable類型的變量的引用是否為DISPOSED
//就可以判斷這個連接器是否中斷
return d == DISPOSED;
}

public static boolean dispose(AtomicReference<Disposable> field) {
Disposable current = field.get();
Disposable d = DISPOSED;
if (current != d) {
//把field設置為DISPOSED
current = field.getAndSet(d);
if (current != d) {
if (current != null) {
current.dispose();
}
return true;
}
}
return false;
}
…
}

可以看到DisposableHelper是個枚舉類，並且只有一個值DISPOSED。dispose方法就是把一個原子引用的field設為DISPOSED，這就是中斷狀態。而isDisposed()就是根據這個標志來判斷是否中斷的。

再回過頭來看CreateEmiiter類的onNext這些方法

@Override
public void onNext(T t) {
//省略無關代碼

if (!isDisposed()) {
//如果沒有dispose()，才會調用onNext()
observer.onNext(t);
}
}

@Override
public void onError(Throwable t) {
if (!tryOnError(t)) {
//如果dispose()了，會調用到這裏，即最終會崩潰
RxJavaPlugins.onError(t);
}
}

@Override
public boolean tryOnError(Throwable t) {
//省略無關代碼
if (!isDisposed()) {
try {
//如果沒有dispose()，才會調用onError()
observer.onError(t);
} finally {
//onError()之後會dispose()
dispose();
}
//如果沒有dispose()，返回true
return true;
}
//如果dispose()了，返回false
return false;
}

@Override
public void onComplete() {
if (!isDisposed()) {
try {
//如果沒有dispose()，才會調用onComplete()
observer.onComplete();
} finally {
//onComplete()之後會dispose()
dispose();
}
}
}

很容易得出，

• 如果沒有dispose，observer的onNext才會被調用
• onError與onComplete方法互斥，只能其中一個調用到，因為調用其中一個，就會把連接切斷，dispose
• 先onError後onComplete中是onComplete不會被調用，反過來的話，就會崩潰，因為onError中拋出了异常，實際上，dispose了後調用onError都會崩

再看一下操作符Map

public final <R> Observable<R> map(Function<? super T, ? extends R> mapper) {
ObjectHelper.requireNonNull(mapper, “mapper is null”);
return RxJavaPlugins.onAssembly(new ObservableMap<T, R>(this, mapper));
}

public final class ObservableMap<T, U> extends AbstractObservableWithUpstream<T, U> {
final Function<? super T, ? extends U> function;

public ObservableMap(ObservableSource<T> source, Function<? super T, ? extends U> function) {
super(source);
this.function = function;
}

@Override
public void subscribeActual(Observer<? super U> t) {
source.subscribe(new MapObserver<T, U>(t, function));
}

static final class MapObserver<T, U> extends BasicFuseableObserver<T, U> {
final Function<? super T, ? extends U> mapper;

MapObserver(Observer<? super U> actual, Function<? super T, ? extends U> mapper) {
super(actual);
this.mapper = mapper;
}

@Override
public void onNext(T t) {
if (done) {
return;
}

if (sourceMode != NONE) {
downstream.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;
}
downstream.onNext(v);
}

…
}
}

可以看到，操作符其實和上面分析的特殊情况下的一樣的，這裏就省略分析了。

##三.Rxjava線程切換
我們一般是這麼使用的

Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onComplete();
}
}).map(new Function<Integer, Integer>() {
@Override
public Integer apply(Integer integer) throws Exception {
return integer+1;
}
}).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<Integer>() {
…
});

通過subscribeOn來切換上遊線程，observeOn來切換下遊線程。

那麼在源碼中，是怎麼的呢？

subscribeOn源碼分析
Schedulers.io()

subscribeOn類型有好幾種，這裏就隨便選擇了Schedulers.io()來分析，別的其實都差不多的，分析了一個就行了。

@NonNull
public static Scheduler io() {
//又是hook，就相當於IO
return RxJavaPlugins.onIoScheduler(IO);
}

public final class Schedulers {
…
@NonNull
static final Scheduler IO;

…

static final class IoHolder {
static final Scheduler DEFAULT = new IoScheduler();
}

static {
//又是hook，就相當於new IOTask
IO = RxJavaPlugins.initIoScheduler(new IOTask());
…
}

…
static final class IOTask implements Callable<Scheduler> {
@Override
public Scheduler call() throws Exception {
return IoHolder.DEFAULT;
}
}
}

可以看到，最後這裏就相當於new IoScheduler，先不看它的具體實現。

subscribeOn

我們繼續看subscribeOn的源碼

public final Observable<T> subscribeOn(Scheduler scheduler) {
ObjectHelper.requireNonNull(scheduler, “scheduler is null”);
return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this, scheduler));
}

可以看到和前面一樣，就相當於返回new 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是為了創建連接器，以便以後切斷等控制的，可以只看裏面
parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
}
…
}

它的構造就是把source和scheduler兩個都保存一下，在後面要用到的。

接下來我們來看訂閱過程，雖然這裏是線程切換，但是其實它也只是個操作符，和我們前面分析的是一樣的，訂閱過程和上面也是一樣的，所以我們可以知道，當訂閱發生後，ObservableSubscribeOn的subscribeActual方法就會被調用。

同樣的，subscribeActual方法中，它把我們自定義的下遊觀察者observer包裝成了SubscribeOnObserver對象，然後調用observer的onSubscribe方法，可以看到，目前為止，還沒有發生任何的線程相關的東西，所以observer的onSubscribe()方法是運行在當前線程中的，那我們重點來看一下parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));方法。

我們先來看一下SubscribeTask類

//是ObservableSubscribeOn的內部類，實現runnable接口，看到這，我們嗅到了線程的味道
final class SubscribeTask implements Runnable {
private final SubscribeOnObserver<T> parent;

SubscribeTask(SubscribeOnObserver<T> parent) {
this.parent = parent;
}

@Override
public void run() {
//這是的source就是我們自定義的Observable對象，就是ObservableCreate
source.subscribe(parent);
}
}

可以看到，這個類非常簡單，實現了Runnable接口，在run方法中調用source.subscribe(parent);，這是個鏈式調用，會一層一層調用上去。

再來看scheduler.scheduleDirect

這是線程切換的核心部分了，一定要仔細看

public Disposable scheduleDirect(@NonNull Runnable run) {
return scheduleDirect(run, 0L, TimeUnit.NANOSECONDS);
}

//run就是SubscribeTask
public Disposable scheduleDirect(@NonNull Runnable run, long delay, @NonNull TimeUnit unit) {
//createWorker在Schedule類中是個抽象方法，所以實現是在子類中
//所以這個方法就是在IOSchedule中實現的
//worker中可以執行runnabale
final Worker w = createWorker();

//實際上decoratedRun還是個run對象，也就是SubscribeTask
final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);

//runnable和worker包裝成一個DisposeTask
DisposeTask task = new DisposeTask(decoratedRun, w);

//Worker執行這個Task
w.schedule(task, delay, unit);

return task;
}

上面的代碼注釋已經寫得非常詳細了，scheduleDirect方法就是，new一個worker，然後使用這個worker來執行task線程。

再看一下IoIoScheduler中，createWorker以及shedule的過程

public Worker createWorker() {
//new一個EventLoopWorker並傳一個worker的緩存池進去
return new EventLoopWorker(pool.get());
}

static final class EventLoopWorker extends Scheduler.Worker {
private final CompositeDisposable tasks;
private final CachedWorkerPool pool;
private final ThreadWorker threadWorker;

final AtomicBoolean once = new AtomicBoolean();

EventLoopWorker(CachedWorkerPool pool) {
this.pool = pool;
this.tasks = new CompositeDisposable();
//從緩存worker池中取一個worker出來
this.threadWorker = pool.get();
}

…

@NonNull
@Override
public Disposable schedule(@NonNull Runnable action, long delayTime, @NonNull TimeUnit unit) {
if (tasks.isDisposed()) {
// don’t schedule, we are unsubscribed
return EmptyDisposable.INSTANCE;
}

//Runnable交給threadWorker去執行
return threadWorker.scheduleActual(action, delayTime, unit, tasks);
}
}

注意的是，不同的Scheduler類會有不同的Worker實現，因為Scheduler類最終都是交由worker來執行調度的，不過分析起來差別不大。

接下來我們看worker的緩存池操作

static final class CachedWorkerPool implements Runnable {
…

ThreadWorker get() {
if (allWorkers.isDisposed()) {
return SHUTDOWN_THREAD_WORKER;
}
while (!expiringWorkerQueue.isEmpty()) {
//緩沖池不為空，就從緩存池中取一個threadWorker
ThreadWorker threadWorker = expiringWorkerQueue.poll();
if (threadWorker != null) {
return threadWorker;
}
}

// No cached worker found, so create a new one.
//為空就一個並返回去
ThreadWorker w = new ThreadWorker(threadFactory);
allWorkers.add(w);
return w;
}
…
}

再看worker的執行代碼threadWorker.scheduleActual

代碼跟進，會發現實現在它的父類NewThreadWorker中

public class NewThreadWorker extends Scheduler.Worker implements Disposable {
private final ScheduledExecutorService executor;

volatile boolean disposed;

public NewThreadWorker(ThreadFactory threadFactory) {
//在構造中創建一個ScheduledExecutorService對象
//可以通過它來使用線程池
executor = SchedulerPoolFactory.create(threadFactory);
}
…
public ScheduledRunnable scheduleActual(final Runnable run, long delayTime, @NonNull TimeUnit unit, @Nullable DisposableContainer parent) {
//這是decoratedRun就相當於run
Runnable decoratedRun = RxJavaPlugins.onSchedule(run);

//將decoratedRun包裝成一個新對象ScheduledRunnable
ScheduledRunnable sr = new ScheduledRunnable(decoratedRun, parent);

if (parent != null) {
if (!parent.add(sr)) {
return sr;
}
}

Future<?> f;
try {
if (delayTime <= 0) {
//線程池中立即執行ScheduledRunnable
f = executor.submit((Callable<Object>)sr);
} else {
//線程池中延遲執行ScheduledRunnable
f = executor.schedule((Callable<Object>)sr, delayTime, unit);
}
sr.setFuture(f);
} catch (RejectedExecutionException ex) {
…
}

return sr;
}
…
}

這裏的executor就是使用線程池來執行任務，最終subscribeTask的run方法會在線程池中被執行，即上遊的Observable的subscribe方法會在IO線程中調用了。

小結

• Observer的onSubscribe方法運行在當前線程中，因為源碼中並沒有線程切換
• 如果設置了subscribeOn(指定線程)，那麼Observable中的subscribe方法將會運行在指定線程中。
• 當多個subscribeOn調用時，因為從源碼可知，線程的切換是從下往上的，最後也就是鏈式調用的第一個切換過程，才是有效的切換

observeOn源碼分析

.observeOn(AndroidSchedulers.mainThread())

AndroidSchedulers.mainThread()
同樣的，我們先看AndroidSchedulers.mainThread()的源碼

public static Scheduler mainThread() {
return RxAndroidPlugins.onMainThreadScheduler(MAIN_THREAD);
}
private static final Scheduler MAIN_THREAD = RxAndroidPlugins.initMainThreadScheduler(
new Callable<Scheduler>() {
@Override public Scheduler call() throws Exception {
return MainHolder.DEFAULT;
}
});
private static final class MainHolder {
static final Scheduler DEFAULT
= new HandlerScheduler(new Handler(Looper.getMainLooper()), false);
}

這一段代碼相信如果是看了上面的源碼分析的話，一眼就能看出來，其實就相當於new HandlerScheduler(new Handler(Looper.getMainLooper()), false);，把一個主線程的Handler包裝進了HandlerScheduler中。

observeOn
然後我們繼續看observeOn的源碼

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));
}

通過源碼也可以知道，這裏相當於直接new ObservableObserveOn

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) {
//是當前線程的話，直接調用下遊的subscribe方法
//也就是調用下一個Observable的subscibe方法
source.subscribe(observer);
} else {
//創建worker
//本例中的schedule為HandlerScheduler
Scheduler.Worker w = scheduler.createWorker();

//這裏和上面分析有點類似，會將worker包裝到ObserveOnObserver中
//注意：source.subscribe沒有涉及到worker，所以還是在之間設置的線程中執行
source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
}
}
…
}

首先，判斷是否已經在要切換的線程上了，如果是的話，那麼直接調用。如果不是，那麼使用HandlerScheduler包裝一下worker，然後通過worker來把下遊的事件進行切換，直接上遊訂閱，不做線程操作。

我們來看ObserveOnObserver類的源碼

static final class ObserveOnObserver<T> extends BasicIntQueueDisposable<T>
implements Observer<T>, Runnable {
…

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 onNext(T t) {
if (done) {
return;
}

if (sourceMode != QueueDisposable.ASYNC) {
//將信息存入隊列中
queue.offer(t);
}
schedule();
}
…

void schedule() {
if (getAndIncrement() == 0) {
//在這裏調用
worker.schedule(this);
}
}

void drainNormal() {
int missed = 1;

//存儲消息的隊列
final SimpleQueue<T> q = queue;
//這裏的downstram實際就是下遊的observer
final Observer<? super T> a = downstream;

for (; {
if (checkTerminated(done, q.isEmpty(), a)) {
return;
}

for (; {
boolean d = done;
T v;

try {
//從隊列中取出消息
v = q.poll();
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
disposed = true;
upstream.dispose();
q.clear();
a.onError(ex);
worker.dispose();
return;
}
boolean empty = v == null;

if (checkTerminated(d, empty, a)) {
return;
}

if (empty) {
break;
}

//調用下遊的Observer的onNext
a.onNext(v);
}

missed = addAndGet(-missed);
if (missed == 0) {
break;
}
}
}

@Override
public void run() {
//outputFused默認是false
if (outputFused) {
drainFused();
} else {
//所以默認調用drainNormal
drainNormal();
}
}
…
}

由上面鏈式調用的分析可以知道，source.subscribe(observer)被調用時，會調用它

CreateEmitter<T> parent = new CreateEmitter<T>(observer);
observer.onSubscribe(parent);

try {
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}

可以知道，會通過發射器把上遊的事件都發送到下遊，然後訂閱，所以ObserveOnObserver這個類中，onNext這個方法就會調用，然後執行schedule，最後執行worker.schedule(this);，因為傳入的runnable是this，也就是最後線程後調用到本類的run方法，執行run，最終執行drainNormal()方法。
那我們再來看一下worker中是怎麼create以及調用的。
在上面我們已經知道，這個schedule是HandlerSchedule

final class HandlerScheduler extends Scheduler {
private final Handler handler;
private final boolean async;

HandlerScheduler(Handler handler, boolean async) {
this.handler = handler;
this.async = async;
}

…

@Override
public Worker createWorker() {
//主線程的handler
return new HandlerWorker(handler, async);
}
}

接著看HandlerWorker的schedule方法

private static final class HandlerWorker extends Worker {
private final Handler handler;
private final boolean async;

private volatile boolean disposed;

HandlerWorker(Handler handler, boolean async) {
this.handler = handler;
this.async = async;
}

@Override
@SuppressLint(“NewApi”) // Async will only be true when the API is available to call.
public Disposable schedule(Runnable run, long delay, TimeUnit unit) {
if (run == null) throw new NullPointerException(“run == null”);
if (unit == null) throw new NullPointerException(“unit == null”);

if (disposed) {
最後為了幫助大家深刻理解Android相關知識點的原理以及面試相關知識，這裏放上相關的我搜集整理的24套騰訊、字節跳動、阿裏、百度2020-2021面試真題解析，我把技術點整理成了視頻和PDF（實際上比預期多花了不少精力），包知識脈絡 + 諸多細節。

還有?高級架構技術進階腦圖、Android開發面試專題資料?幫助大家學習提昇進階，也節省大家在網上搜索資料的時間來學習，也可以分享給身邊好友一起學習。

點擊：

《 Android架構視頻+BAT面試專題PDF+學習筆記》即可免費獲取~

網上學習 Android的資料一大堆，但如果學到的知識不成體系，遇到問題時只是淺嘗輒止，不再深入研究，那麼很難做到真正的技術提昇。希望這份系統化的技術體系對大家有一個方向參考。