12.21 高級架構師之路---RxJava原理分析

一、什麼是RXJAVA

RXJAVA是一個庫，用來支持我們需求裡需要異步操作的地方

二、RXJAVA的原理

RXJAVA的實現，是一種擴展式的觀察者模式。

RXJAVA中有四種概念。observable（被觀察者），observer（觀察者），subscribe（訂閱），事件。Observable和Observer通過subscribe來實現訂閱關係。與傳統的觀察者模式不同，除了onNext事件外，Rxjava還提供了onCompleted和onError。當不再有onNext事件發送時，將以onCompleted事件作為結束。當處理過程中出現異常時，會觸發onError，同時隊列自動終止，不允許再有事件發出。onCompleted和onError在一個序列中有且只有一個，二者互斥，只能出現一個。

subscribeOn和observeOn

subscribeOn調用可以將之前的操作加如線程池，從而保證運行於子線程中，observeOn會使後邊的執行運行於主線程，這裡的之前和後邊均是指的代碼結構上的前後。

經過分析可知道，當subscribeOn調用的時候，會創建一個ObservableSubscribeOn對象返回，與此同時，上一級產生的對象會被保存在當前對象的source變量中，並且，將創建出一個線程池，先看線程池的創建，這裡直接以io線程為例
Schedulers.io(Schedulers.io())

<code>public static Scheduler io() {
       return RxJavaPlugins.onIoScheduler(IO);
}/<code>

其中的IO是在Schedulers類加載的時候就創建出來的，從這個結構可以看出，IO就是IoScheduler對象，RxJavaPlugins.initIoScheduler方法接收一個Callable線程，返回callable.call，也就是call方法中返回的就是這個函數的返回值（Callable是另一種開啟線程的方式，這個線程有返回值，當返回值獲取到之前，會阻塞當前線程）

<code>IO = RxJavaPlugins.initIoScheduler(new Callable<scheduler>() {
            @Override
            public Scheduler call() throws Exception {
                return IoHolder.DEFAULT;
            }
});

static final class IoHolder {
        static final Scheduler DEFAULT = new IoScheduler();
}/<scheduler>/<code>

那麼IoScheduler是什麼？當IoScheduler創建的時候

<code>public IoScheduler() {
      this.pool = new AtomicReference<cachedworkerpool>(NONE);
      start();
}
@Override
public void start() {
     CachedWorkerPool update = new CachedWorkerPool(KEEP_ALIVE_TIME, KEEP_ALIVE_UNIT);
     if (!pool.compareAndSet(NONE, update)) {
         update.shutdown();
     }
}/<cachedworkerpool>/<code>

NONE是IoScheduler中創建的一個線程池,所以IoScheduler其實就是一個封裝好了的線程池對象

<code>static final CachedWorkerPool NONE;
static 
     NONE = new CachedWorkerPool(0, null);
}
CachedWorkerPool(long keepAliveTime, TimeUnit unit) {
     this.keepAliveTime = unit != null ? unit.toNanos(keepAliveTime) : 0L;
     this.expiringWorkerQueue = new ConcurrentLinkedQueue<threadworker>();
     this.allWorkers = new CompositeDisposable();
     ScheduledExecutorService evictor = null;
     Future> task = null;
     if (unit != null) {
        evictor = Executors.newScheduledThreadPool(1, EVICTOR_THREAD_FACTORY);
        task = evictor.scheduleWithFixedDelay(this, this.keepAliveTime, this.keepAliveTime, TimeUnit.NANOSECONDS);
     }
     evictorService = evictor;
     evictorTask = task;
}/<threadworker>/<code>

Schedulers.io(Schedulers.io())的調用，執行了兩個動作，第一，保存上一級的對象，第二創建線程池

observeOn(AndroidSchedulers.mainThread())

接下來來看主線程的切換,調用observeOn方法，創建ObservableObserveOn對象，同樣保存上一級產生的對象到source中，這裡指的就是subscribeOn返回的對象ObservableSubscribeOn，並且保存傳入的Scheduler--AndroidSchedulers.mainThread()

<code>@CheckReturnValue
@SchedulerSupport(SchedulerSupport.CUSTOM)
public final Observable observeOn(Scheduler scheduler) {
    return observeOn(scheduler, false, bufferSize());
}
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.CUSTOM)
public final Observable observeOn(Scheduler scheduler, boolean delayError, int bufferSize) {
    ObjectHelper.requireNonNull(scheduler, "scheduler is null");
    ObjectHelper.verifyPositive(bufferSize, "bufferSize");
    return RxJavaPlugins.onAssembly(new ObservableObserveOn(this, scheduler, delayError, bufferSize));
}/<code>

進入AndroidSchedulers.mainThread(),與上邊同樣的寫法，最後返回HandlerScheduler

<code>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 {
        //可以猜測這個HandlerScheduler是一個通過對Handler進行封裝
        //運行於主線程的線程，可以看到Looper.getMainLooper()傳入了一個主線程的
        //looper對象，事實上也是如此
        static final Scheduler DEFAULT = new HandlerScheduler(new Handler(Looper.getMainLooper()));
    }
}/<scheduler>/<code>

所以，很類似，observeOn(AndroidSchedulers.mainThread())同樣是做了兩件事，保存source和Scheduler，那麼兩種線程是如何進行調度的，其實看到這裡，還沒有進入正題，真正的邏輯其實在subscribe方法上。

subscribe

以subscribe(new Observer<string>())為例說明（new Consumer最終源碼也是相同的），調用subscribe方法後會來到Observable的抽象方法subscribeActual中，所以我們要到當前Observable實現類中找這個方法，按照上邊程序調用的順序，此時，調用subscribe方法的對象是observeOn方法產生的ObservableObserveOn，進入這個類，找到subscribeActual方法/<string>

<code>@Override
protected void subscribeActual(Observer super T> observer) {
    //這個scheduler是指AndroidSchedulers.mainThread()，也就是HandlerScheduler
    if (scheduler instanceof TrampolineScheduler) {
         source.subscribe(observer);
    } else {
         //創建一個worker
         Scheduler.Worker w = scheduler.createWorker();
         source.subscribe(new ObserveOnObserver(observer, w, delayError, bufferSize));
    }
}/<code>

進入HandlerScheduler找到createWorker方法，這裡創建了一個HandlerWorker對象,看到這裡大概也可以猜測一下，HandlerWorker中的schedule方法將會是一個關鍵，傳入的handler是主線程中的handler，明顯是要通過消息機制發送到主線程執行，問題的關鍵，在於是怎麼發送到主線程執行的，schedule方法的具體執行我們暫且不看，按照程序執行順序繼續往下走

<code>@Override
public Worker createWorker() {
    return new HandlerWorker(handler);
}

private static final class HandlerWorker extends Worker {
    private final Handler handler;
    private volatile boolean disposed;
    HandlerWorker(Handler handler) {
         this.handler = 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, unit.toMillis(delay));
         ......
         return scheduled;
    }
         ......
}/<code>

在創建了worker之後，調用方法subscribe,source很明顯是ObservableObserveOn對象創建的時候所保存的上一級的調用subscribeOn方法產生的ObservableSubscribeOn對象，通過這個對象調用subscribe方法，又會進入到ObservableSubscribeOn的subscribeActual方法。observer指的是我們代碼中傳入的observer（subscribe時new的那個），這裡對observer封裝了一層，以ObserveOnObserver的形式傳入到ObservableSubscribeOn的subscribeActual方法中，向上層傳遞了一級，可以參考08.RxJava運作流程源碼分析中提供的流程圖

<code>//source指ObservableSubscribeOn對象
source.subscribe(new ObserveOnObserver(observer, w, delayError, bufferSize));/<code>

來到ObservableSubscribeOn的subscribeActual

<code>@Override
//參數s指的時對observer封裝了一層之後的ObserveOnObserver（new ObserveOnObserver(new Observer )） 

public void subscribeActual(final Observer super T> s) {
    //對ObserveOnObserver對象進行一次封裝
    //此時Observer已經被封裝了兩層
    //（new SubscribeOnObserver(new ObserveOnObserver(new Observer))）
    final SubscribeOnObserver parent = new SubscribeOnObserver(s);
    //調用ObserveOnObserver對象的onSubscribe
    s.onSubscribe(parent);
    parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
}/<code>

看看ObserveOnObserver的onSubscribe方法

<code>@Override
public void onSubscribe(Disposable s) {
    if (DisposableHelper.validate(this.s, s)) {
       this.s = s;
       //注意這裡這個判斷這次是不會滿足的，也就是這裡的代碼不會走
       if (s instanceof QueueDisposable) {
           @SuppressWarnings("unchecked")
           QueueDisposable qd = (QueueDisposable) s;

           int m = qd.requestFusion(QueueDisposable.ANY | QueueDisposable.BOUNDARY);
           //同步，如果是要同步執行，就是指如果設置了在主線程執行，那麼
           //就執行schedule()，往下看可以發現是使用我我們創建的worker
           //發送到主線程執行
           if (m == QueueDisposable.SYNC) {
              sourceMode = m;
              queue = qd;
              done = true;
              //actual指的就是我們傳入的最原始的那個observer
              actual.onSubscribe(this); 

              schedule();
              return;
            }
            //異步，如果是異步執行，直接在當前線程執行，當前線程也就是子線程
            if (m == QueueDisposable.ASYNC) {
               sourceMode = m;
               queue = qd;
               actual.onSubscribe(this);
               return;
            }

                queue = new SpscLinkedArrayQueue(bufferSize);
                //actual是我們new的那個Observer，所以這裡直接回調了onSubscribe方法
                actual.onSubscribe(this);
            }
        }/<code>

scheduler就是Schedulers.io()得到的就是IoSchedule對象，在上邊分析subscribeOn方法時我們已經知道這個對象是一個線程池，調用scheduleDirect方法就是將SubscribeTask這個Runnable放進了線程池執行，並且是在子線程中

<code>@NonNull
public Disposable scheduleDirect(@NonNull Runnable run, long delay, @NonNull TimeUnit unit) {
    final Worker w = createWorker();
    final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
    DisposeTask task = new DisposeTask(decoratedRun, w);
    w.schedule(task, delay, unit);
    return task;
}/<code>

createWorker()是個抽象類，在IoSchedule中找到重寫的方法

<code>@Override
public Worker createWorker() { 

    return new EventLoopWorker(pool.get());
}/<code>

所以這樣一來也就是說new SubscribeTask(parent))這個Runnable被放入了線程池執行，這時候會調用它的run方法,這樣就又回到了調用上一級產生對象的subscribe方法中去了，不同的是此時subscribe已經是在線程池中執行了（子線程）

<code>@Override
public void run() {
    source.subscribe(parent);
}/<code>

就這樣一級一級的往上調用，下一個會走到ObservableMap的subscribeActual方法，最後走到ObservableJust的subscribeActual,s.onSubscribe(sd)方法並沒有執行什麼東西，onSubscribe在之前已經被調用了，重點在 sd.run()

<code>@Override
protected void subscribeActual(Observer super T> s) {
    ScalarDisposable sd = new ScalarDisposable(s, value);
    s.onSubscribe(sd);
    sd.run();
}/<code>

終於在這裡要看到onNext onComplete方法的執行了

<code>@Override
public void run() {
    if (get() == START && compareAndSet(START, ON_NEXT)) {
        //observer 是 new MapObserver(new SubscribeOnObserver(new ObserveOnObserver(new Observer(){......})))
        observer.onNext(value);
        if (get() == ON_NEXT) { 

            lazySet(ON_COMPLETE);
            observer.onComplete();
        }
    }
}/<code>

此時一層一層的調用到這裡，observer對象已經是經過層層封裝包裹的observer了(new MapObserver(new SubscribeOnObserver(new ObserveOnObserver(new Observer(){......})))),所以調用observer.onNext會首先執行MapObserver中的onNext,不管用戶調用了幾次map操作符，都會一個一個的通過回調onNext方法執行完成（如果有多個map方法被調用，當執行完一個apply方法後，後邊的actual.onNext就會進入下一個MapObserver中的onNext方法），當執行到最後一個onNext方法的時候，此時這個actual表示的就是SubscribeOnObserver對象了，也就會去執行它裡邊的onNext

<code>@Override
public void onNext(T t) {
     if (done) {
        return;
     }
     if (sourceMode != NONE) {
        actual.onNext(null);
        return;
     }
     U v;
     try {
         //執行apply方法，也就是map操作符中的回調方法
         v = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper function returned a null value.");
     } catch (Throwable ex) {
         fail(ex);
         return;
     }
     actual.onNext(v);
}/<code>

SubscribeOnObserver中的onNext，這裡的actual指的是ObserveOnObserver，所以又要去執行它的onNext

<code>@Override
public void onNext(T t) {
   actual.onNext(t);
}/<code>

ObserveOnObserver中的onNext

<code>@Override
public void onNext(T t) {
    if (done) {
       return;
    }
    if (sourceMode != QueueDisposable.ASYNC) {
       queue.offer(t);
    }
    schedule();
}

void schedule() {
    if (getAndIncrement() == 0) {
    //這個worker是AndroidScheduler.mainThread得到的一個運行於主線程的封裝類 HandlerWorker         
    worker.schedule(this);
    }
}/<code>

在分析observeOn方法的時候我們已經知道這個worker是AndroidScheduler.mainThread得到的一個運行於主線程的封裝類 HandlerWorker ，worker.schedule(this)傳入的是一個Runnable，也就是會在主線程中執行這個Runnable，我們找到重寫的run方法。終於找到onNext和onComplete的最終執行的地方了，並且我們知道，這兩個方法是在主線程執行的

<code>@Override
public void run() {
     if (outputFused) {
        drainFused();
     } else {
        //會執行這個，上邊那個先不管
        drainNormal(); 

     }
}
void drainNormal() {
     int missed = 1;
     final SimpleQueue 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);
           }
           missed = addAndGet(-missed);
           if (missed == 0) {
                break;
           }
      }
}/<code>

到這裡，RxJava線程調度的實現方式基本上我們已經瞭解了。
這裡可以插一個題外話，通常我們使用handler發送的消息都是在handleMessage方法中執行，但是這裡我們無論如何找不到這個方法的實現，那麼handler是如何處理消息的？

<code>@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, unit.toMillis(delay));
    ......
    return scheduled;
}/<code>

可以看到這裡Message message = Message.obtain(handler, scheduled)，看一下obtain方法的源碼會發現傳入的第二個參數是一個callback，保存到了message的成員變量m.callback中，當handler調用sendMessageDelayed會將消息加入主線程的消息隊列（因為handler就是主線程的handler），我們知道應用啟動就會初始化一個主線程的handler一個looper和messageQueue（對消息機制不理解的可以看另一篇15.源碼閱讀（安卓消息機制）），調用looper.loop開啟一個無限循環不斷的從主線程消息隊列中取消息，我們看看它是如何取的

<code>public static void loop() {
    for (;;) {
        Message msg = queue.next(); // might block
        ......
        msg.target.dispatchMessage(msg);
    }
}/<code>

無限循環中取到message後會執行發送這個Message的handler中的dispatchMessage方法,這時候會判斷callback也就是我們上邊那個傳入的，如果它不能與null,就執行handleCallback，執行callback的run方法，找到這裡終於找到為什麼沒有handlerMessage仍然可以處理消息了

<code>public void dispatchMessage(Message msg) {
        if (msg.callback != null) { 

            handleCallback(msg);
        } else {
            if (mCallback != null) {
                if (mCallback.handleMessage(msg)) {
                    return;
                }
            }
            handleMessage(msg);
        }
}
private static void handleCallback(Message message) {
      message.callback.run();
}/<code>

傳入的callback是哪個，就是 Message message = Message.obtain(handler, scheduled)中的schedule，schedule是哪個ScheduledRunnable ，也就是說執行的是ScheduledRunnable 的run方法,delegate就是ScheduledRunnable 中傳入的那個runnable，追溯上去，這個runnable就是worker.schedule(this)中的this，所以可以找到重寫的run方法

<code>@Override
public void run() {
      try {
           delegate.run();
      } catch (Throwable t) {
           RxJavaPlugins.onError(t);
      }
}/<code>

專注於技術熱點大數據，人工智能，JAVA、Python、 C 、GO、Javascript等語言最新前言技術，及業務痛點問題分析，請關注【編程我最懂】共同交流學習。

閱讀更多 編程我最懂 的文章

關鍵字: 架構師 return 之路