Rxjava（二)线程切换研究

使用过Rxjava的都知道，Rxjava 通过subscribeOn() 和observerOn()来控制线程的切换，在RxJava事件的切换研究中我们知道Rxjava逻辑块，分为构建和调用了subscribe
后执行过程，subscribeOn()就是控制构建过程的，视觉上的构建就是下图上面框里这部分线程，observerOn控制最后发起执行的部分的线程

image3.png

那么Rxjava怎么实现线程的调度呢？
起码通过RxJava事件的切换研究学习，我们立马想到，要想维持流式语法，除了最后一部分subscribe，上面无论加什么操作最后肯定返回的是
一个Observable的实现类，subscribeOn() 和observerOn()也是如此，看简化后的源码

public abstract class Observable<T> implements ObservableSource<T>{

    public static <T> Observable<T> just(T item) {
        return onAssembly(new ObservableJust<T>(item));
    }

    private static <T> Observable<T> onAssembly(Observable<T> source) {
        // 留出来了
        return source;
    }

    @Override
    public void subscribe(Observer<T> observer) {
        subscribeActual(observer);
    }

    public void subscribe(Consumer<T> onNext){
        subscribe(onNext,null,null);
    }

    private void subscribe(Consumer<T> onNext, Consumer<T> error, Consumer<T> complete) {
        subscribe(new LambdaObserver<T>(onNext));
    }

    protected abstract void subscribeActual(Observer<T> observer);

    public <R> Observable<R> map(Function<T, R> function) {
        return onAssembly(new ObservableMap<>(this,function));
    }

    public Observable<Bitmap> subscribeOn(Schedulers schedulers) {
        return onAssembly(new ObservableSchedulers(this,schedulers));
    }

    public Observable<T> observerOn(Schedulers schedulers) {
        return onAssembly(new ObserverOnObservable(this,schedulers));
    }
}

怎么样，确实如此，又是调了onAssembly方法，最后返回新的Observalbe对象

下一步呢，不同于分析过的map操作，第二个参数不再是是function,而是Schedulers，我们先进去看看ObservaleScheduler,

final class ObservableSchedulers<T> extends Observable<T> {
    final Observable<T> source;
    final Schedulers schedulers;

    public ObservableSchedulers(Observable<T> source, Schedulers schedulers) {
        this.source = source;
        this.schedulers = schedulers;
    }

    @Override
    protected void subscribeActual(Observer<T> observer) {
        schedulers.scheduleDirect(new SchedulerTask(observer));
    }

    private class SchedulerTask implements Runnable{
        final Observer<T> observer;
        public SchedulerTask(Observer<T> observer) {
            this.observer = observer;
        }

        @Override
        public void run() {
            // 线程池最终回来执行 Runnable -> 这行代码，会执行上游的 subscribe()
            // 而这个run方法在子线程中
            source.subscribe(observer);
        }
    }
}

不同于map操作符的Observable的subscribeActual 直接调用source.subscribe(observer)
ObservaleScheduler中的source.subscribe(observer)包裹在一个新的runnable线程中执行，那么实例化时传入的schedulers也就时chedulers.io()是什么呢？
想也不用想，肯定是线程池

public abstract class Schedulers {
    static Schedulers MAIN_THREAD;
    static Schedulers IO;
    static {
        IO = new IOSchedulers();
        MAIN_THREAD = new MainSchedulers(new Handler(Looper.getMainLooper()));
    }

    public static Schedulers io() {
        return IO;
    }

    public abstract void scheduleDirect(Runnable runnable);

    public static Schedulers mainThread() {
        return MAIN_THREAD;
    }

    private static class IOSchedulers extends Schedulers {
        ExecutorService service;
        public IOSchedulers(){
            service = Executors.newScheduledThreadPool(1, new ThreadFactory() {
                @Override
                public Thread newThread(@NonNull Runnable r) {
                    return new Thread(r);
                }
            });
        }

        @Override
        public void scheduleDirect(Runnable runnable) {
            service.execute(runnable);
        }
    }

    private static class MainSchedulers extends Schedulers {
        private Handler handler;
        public MainSchedulers(Handler handler) {
            this.handler = handler;
        }

        @Override
        public void scheduleDirect(Runnable runnable) {
            Message message = Message.obtain(handler,runnable);
            handler.sendMessage(message);
        }
    }
}

怎么样，这下大概明白了，线程调度操作符区别于其他的操符，就是通过处理，在执行流程阶段，subscirbeActual()中对应加入了线程

总结一下Rxjava的两篇文章，其实Rxjava核心的编码就是每个逻辑过程中的操作符，他们都有自己的一个Observable实现类，和Observer,例如map的
ObservableMap和ObserverMap,在构建阶段，都是通过onAssemby()方法的this参数，持有了上游的Observable，而在执行向上的阶段，通过subscribeActual
方法，实际执行上游的subscribe()把自己的observer传给上游，最后到了顶层的Observable，一层一层调用下级的Observer的OnNext实现整个流程

在线程切换中，只不过在subscribeActual执行时，包裹在线程中```