1. 创建线程池和线程管理策略分析

// 在开发中使用Rxjava来完成线程切换会调用到以下方法(还有几个就不一一列举了，原理一样的)，那么就从这里开始分析

Schedulers.io()

Schedulers.computation()

Schedulers.newThread()

AndroidSchedulers.mainThread()

当我们调用以上方法中的任意一个，都会调到Schedulers类中，Schedulers使用策略模式封装了所有线程切换策略(因此后面以io()分析)。

// 1. Schedulers类中，静态创建IOTask(),当调用Schedulers.io()的时候，就是返回这个Callable.

static {

SINGLE = RxJavaPlugins.initSingleScheduler(new SingleTask());

COMPUTATION = RxJavaPlugins.initComputationScheduler(new ComputationTask());

IO = RxJavaPlugins.initIoScheduler(new IOTask());

TRAMPOLINE = TrampolineScheduler.instance();

NEW_THREAD = RxJavaPlugins.initNewThreadScheduler(new NewThreadTask());

}

// 2.创建IoScheduler

static final class IOTask implements Callable {

@Override

public Scheduler call() throws Exception {

return IoHolder.DEFAULT;

}

}

static final class IoHolder {

static final Scheduler DEFAULT = new IoScheduler();

}

// 3.创建线程池

public IoScheduler(ThreadFactory threadFactory) {

this.threadFactory = threadFactory;

this.pool = new AtomicReference(NONE);

start();

}

public void start() {

// CachedWorkerPool任务池，里面持有任务队列和线程池

CachedWorkerPool update = new CachedWorkerPool(KEEP_ALIVE_TIME, KEEP_ALIVE_UNIT, threadFactory);

if (!pool.compareAndSet(NONE, update)) {

update.shutdown();

}

}

//4. CachedWorkerPool构造方法中创建线程池，并且暴露get()提供需要执行的任务

static final class CachedWorkerPool implements Runnable {

private final long keepAliveTime;

private final ConcurrentLinkedQueue expiringWorkerQueue;

final CompositeDisposable allWorkers;

private final ScheduledExecutorService evictorService;

private final Future> evictorTask;

private final ThreadFactory threadFactory;

CachedWorkerPool(long keepAliveTime, TimeUnit unit, ThreadFactory threadFactory) {

......

if (unit != null) {

// 创建线程池

evictor = Executors.newScheduledThreadPool(1, EVICTOR_THREAD_FACTORY);

task = evictor.scheduleWithFixedDelay(this, this.keepAliveTime, this.keepAliveTime, TimeUnit.NANOSECONDS);

}

......

}

ThreadWorker get() {

.....

while (!expiringWorkerQueue.isEmpty()) {

// 任务队列不为空，从队列中取一个并返回

ThreadWorker threadWorker = expiringWorkerQueue.poll();

if (threadWorker != null) {

return threadWorker;

}

}

// 如果任务队列是空的，就创建一个并返回

ThreadWorker w = new ThreadWorker(threadFactory);

allWorkers.add(w);

return w;

}

......

}

用一张图可能说明得比较清楚一些。

Schedulers调度过程.png

2. Rxjava上游任务在子线程中执行分析

// 上游线程切换使用过程

Observable.create(new ObservableOnSubscribe() {

@Override

public void subscribe(ObservableEmitter e) throws Exception {

e.onNext("JackOu");

}

})

// ObservableCreate.subscribeOn

.subscribeOn(Schedulers.io())

// ObservableSubscribeOn.subscribe

.subscribe(new Observer() {

......

@Override

public void onNext(String s) {

}

......

});

从上面使用过程的代码看下面的图，分析Rxjava封装任务和抛任务到线程池的过程。

上游任务在线程池执行流程图.png

当我们一订阅(调用subscribe(Observer)方法)的时候，Rxjava将会把上游需要执行的任务和下游的观察者经过层层包裹，包裹好之后，就会得到一个Scheduler.Worker任务对象。当调用发射器的onNext的方式的时候，结合第一小节的图片，ObservableSubscribeOn就会将任务抛到线程池执行，在子线程中执行任务并且返回，从而完成线程切换功能。

3. Rxjava下游任务在主线程中执行分析

3.1 创建AndroidSchedulers.mainThread的过程

如第一节Schedulers的创建流程一样，当调用AndroidSchedulers.mainThread()之后，最终会创建HandlerScheduler。

// 1.创建HandlerScheduler，并且传入MainLooper

public final class AndroidSchedulers {

private static final class MainHolder {

// 创建HandlerScheduler

static final Scheduler DEFAULT = new HandlerScheduler(new Handler(Looper.getMainLooper()));

}

private static final Scheduler MAIN_THREAD = RxAndroidPlugins.initMainThreadScheduler(

new Callable() {

@Override public Scheduler call() throws Exception {

return MainHolder.DEFAULT;

}

});

public static Scheduler mainThread() {

return RxAndroidPlugins.onMainThreadScheduler(MAIN_THREAD);

}

}

// 2.当创建任务的时候，创建HandlerWorker

final class HandlerScheduler extends Scheduler {

private final Handler handler;

HandlerScheduler(Handler handler) {

this.handler = handler;

}

@Override

public Worker createWorker() {

return new HandlerWorker(handler);

}

}

// 3.当执行任务的时候

private static final class HandlerWorker extends Worker {

private final Handler handler;

HandlerWorker(Handler handler) {

this.handler = handler;

}

@Override

public Disposable schedule(Runnable run, long delay, TimeUnit unit) {

......

// 包装任务

run = RxJavaPlugins.onSchedule(run);

ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);

// 创建Message包装任务

Message message = Message.obtain(handler, scheduled);

message.obj = this;

// 发送任务到MainLooper中，该任务就在主线程中执行了

handler.sendMessageDelayed(message, Math.max(0L, unit.toMillis(delay)));

......

return scheduled;

}

}

其实真正将任务放在主线程中执行就是上面三个步骤，但是Rxjava增加了很多其他功能，例如解除订阅(将任务包装在Disposable中)，增加hook功能(在任务外面在包装了ScheduledRunnable)等等，其最内层的本质就是我们需要执行的任务。细化的包裹情况如下图：

主线程执行任务.png

4.多个线程切换，以哪个为准

如下面代码，我们作死得切换线程，那么哪些线程会最终执行我们的任务呢

Observable.create(new ObservableOnSubscribe() {

@Override

public void subscribe(ObservableEmitter e) throws Exception {

e.onNext("JackOu");

}

})

.subscribeOn(Schedulers.io()) // 上游切换，靠近上游的生效

.subscribeOn(Schedulers.newThread())

.subscribeOn(Schedulers.computation())

.observeOn(Schedulers.io())

.observeOn(Schedulers.computation())

.observeOn(AndroidSchedulers.mainThread()) // 下游切换，靠近下游的生效

.subscribe(new Observer() {

......

@Override

public void onNext(String s) {

}

......

});

我们可以从第二节和第三节看出，当我们每调用一次subscribeOn方法，上游就会多包装一层Scheduler，在订阅之后，解包裹的时候越靠近“待执行任务”的subscribeOn越后解包，所以最靠近任务的subscribeOn调用会是最终被执行，也就是最终被执行的线程。

因此我们可以总结得到：

总结一： 在多次调用线程切换的时候，第一次调用subscribeOn的线程切换会是最后执行任务的线程；最后调用observeOn切换的线程会是最后执行的线程。

总结二：从调用关系来看，越靠近上游的线程切换，将是最终执行任务的线程；越靠近下游的线程切换，将是最终执行任务的线程。