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 Override