rxjava源码分析之scheduler

心已赠人 2022-05-14 02:28 230阅读 0赞

> 本文基于1.3.8版本分析 rxjava线程切换

**分析代码如下：**

// 被观察者(事件源)
            Observable<String> stringObservable = Observable.create(new Observable.OnSubscribe<String>() {
                @Override
                public void call(Subscriber<? super String> subscriber) {
                    Log.d(TAG, "call() called with: subscriber = [" + subscriber + "]");
                    Log.e("THREAD==", "call" + Thread.currentThread().getName());
                    subscriber.onNext("嘿嘿1");
                    subscriber.onNext("嘿嘿2");
                    subscriber.onNext("嘿嘿3");
                    subscriber.onCompleted();
                }
            }).subscribeOn(Schedulers.io());//详细分析这一行
    
            //观察者
            Subscriber subscriber = new Subscriber() {
                @Override
                public void onStart() {
                    super.onStart();
                }
    
                @Override
                public void onCompleted() {
    
                }
    
                @Override
                public void onError(Throwable e) {
    
                }
    
                @Override
                public void onNext(Object o) {
    
                }
            };
            //事件订阅
            stringObservable.subscribe(subscriber);

**开始划重点分析：**  
首先我们的任务很简单就是分析上面代码中的如下一行

subscribeOn(Schedulers.io());//详细分析这一行

# Schedulers.io方法分析 #

> 开始Schedulers.io()分析 双击点如此方法

//Schedulers.java文件
    public static Scheduler io() {
            return RxJavaHooks.onIOScheduler(getInstance().ioScheduler);
        }

`RxjavaHooks.onIOScheduler` 方法忽略即可，主要用于`hook`方法调用过程。所以我们分析  
`getInstance().ioScheduler`即可

**getInstance()**分析

//Schedulers.java文件
    public final class Schedulers {
        
    
        /*
            ...其他代码省略
        */
        //以下几个自带scheduler的大家应该不陌生
        private final Scheduler computationScheduler;//用与计算的
        private final Scheduler ioScheduler;//用于io
        private final Scheduler newThreadScheduler;//用于每次创建新线程
        //原子性引用 这块如果不了解还请自行补充知识点
        private static final AtomicReference<Schedulers> INSTANCE = new AtomicReference<Schedulers>();
    
        private static Schedulers getInstance() {
            for (;;) {
                //获取对象
                Schedulers current = INSTANCE.get();
                //第一次为null所以不关心
                 if (current != null) {
                    return current;
                }
                //创建对象  继续分析构造方法
                current = new Schedulers();
                //将current对象放入原子性引用INSTANCE中
                if (INSTANCE.compareAndSet(null, current)) {
                    return current;
                } else {
                    current.shutdownInstance();
                }
            }
        }

**new Schedulers();**分析

//Schedulers.java文件
      private Schedulers() {
            @SuppressWarnings("deprecation")
            RxJavaSchedulersHook hook = RxJavaPlugins.getInstance().getSchedulersHook();
    
            Scheduler c = hook.getComputationScheduler();
            if (c != null) {
                computationScheduler = c;
            } else {
            //这里主要分析io 这个的话道理差不多
                computationScheduler = RxJavaSchedulersHook.createComputationScheduler();
            }
    
            //第一次必然为空
            Scheduler io = hook.getIOScheduler();
    
            if (io != null) {
                ioScheduler = io;
            } else {
                //创建对象 主要分析
                ioScheduler = RxJavaSchedulersHook.createIoScheduler();
            }
    
            Scheduler nt = hook.getNewThreadScheduler();
            if (nt != null) {
                newThreadScheduler = nt;
            } else {
                newThreadScheduler = RxJavaSchedulersHook.createNewThreadScheduler();
            }
        }

**ioScheduler = RxJavaSchedulersHook.createIoScheduler();**分析

//RxJavaSchedulersHook.java
    public class RxJavaSchedulersHook {
    
        private final static RxJavaSchedulersHook DEFAULT_INSTANCE = new RxJavaSchedulersHook();
    
    
        public static Scheduler createComputationScheduler() {
            //分析
            return createComputationScheduler(new RxThreadFactory("RxComputationScheduler-"));
        }

**new RxThreadFactory(“RxComputationScheduler-“)**分析

//RxThreadFactory.java
    public final class RxThreadFactory extends AtomicLong implements ThreadFactory {
        
    
        /*
            可以清楚的看到继承ThreadFactory这个类。此接口只有一个方法newThread（），该
            方法返回一个Runable对象。
         */
    
        private static final long serialVersionUID = -8841098858898482335L;
    
        public static final ThreadFactory NONE = new ThreadFactory() {
            @Override public Thread newThread(Runnable r) {
                throw new AssertionError("No threads allowed.");
            }
        };
        //线程名字的前缀
        final String prefix;
    
        public RxThreadFactory(String prefix) {
            this.prefix = prefix;
        }
    
        @Override
        public Thread newThread(Runnable r) {
            //创建一个线程
            Thread t = new Thread(r, prefix + incrementAndGet());
            //设置为守护线程
            t.setDaemon(true);
            return t;
        }
    }

可见**new RxThreadFactory(“RxComputationScheduler-“)**用此对象的产出线程的全部为守护线程。那么**rxjava**便不会自身线程原因影响**jvm**关闭

此对象返回后作为参数调用`createComputationScheduler`返回一个`scheduler`对象  
**createComputationScheduler(new RxThreadFactory（。。。）);**

//RxJavaSchedulersHook.java
      public static Scheduler createComputationScheduler(ThreadFactory threadFactory) {
            if (threadFactory == null) {
                throw new NullPointerException("threadFactory == null");
            }
            //创建一个**EventLoopsScheduler**对象返回
            //因为返回值为Scheduler，所以EventLoopsScheduler应该继承scheduler
            return new EventLoopsScheduler(threadFactory);
        }

分析**EventLoopsScheduler**的构造方法之前我们先分析这个类再看

> 这个类东西有些多第一次看有些难受

首先看这个对象的一些属性把

//EventLoopsScheduler.java
      class EventLoopsScheduler{
        
        //这个对象是什么？所以我们自然要看看这个类干嘛的，这个类是EventLoopsScheduler的
        //内部类
        static final FixedSchedulerPool NONE = new FixedSchedulerPool(null, 0);
        //用于创建线程的，构造方法传入的 前面刚分析
        final ThreadFactory threadFactory;
        //原子性引用
        final AtomicReference<FixedSchedulerPool> pool;
        //大家简单看一下即可
        static final class FixedSchedulerPool {
        
             /*
             ...部分代码省略
             */
            //PoolWorker是什么，继续向下看看这个类是干嘛的
            final PoolWorker[] eventLoops;
    
            FixedSchedulerPool(ThreadFactory threadFactory, int maxThreads) {
    
                /*
                    先简单看下只需要知道创建了一个数值然后实例化数组对象
                */      
                this.eventLoops = new PoolWorker[maxThreads];
                for (int i = 0; i < maxThreads; i++) {
                    this.eventLoops[i] = new PoolWorker(threadFactory);
                }
            }
    
        }
        //又是内部类= =继承NewThreadWorker ,NewThreadWorker这个类很重要后文继续看
        static final class PoolWorker extends NewThreadWorker {
        
            PoolWorker(ThreadFactory threadFactory) {
                super(threadFactory);
            }
        }
      }

//NewThreadWorker.java
    /*
    继承Worker对象，关于这个对象用的时候在说
    */
    public class NewThreadWorker extends Scheduler.Worker implements Subscription {
        
    
        public NewThreadWorker(ThreadFactory threadFactory) {
            //创建一个线程池，这个线程池支持延时调用和定期调用
            //ScheduledExecutorService是jdk提供的
            //线程池核心线程数量为1，需要线程时threadFactory生成
            //传递的threadFactory就是我们前面分析的，每次生成一个守护线程
            ScheduledExecutorService exec = Executors.newScheduledThreadPool(1, threadFactory);
    
            executor = exec;
        }
    }

现在大家应该对**EventLoopsScheduler**大致的认识。

回到前面分析**EventLoopsScheduler（ThreadFactory threadFactory）** 构造方法

//EventLoopsScheduler.java
    
        //现在大家对这个不陌生把
        //大家可以理解为内部装固定个数的线程池
        //并且此线程池为SchedulerThreadPoolExecutor对象
        static final FixedSchedulerPool NONE = new FixedSchedulerPool(null, 0);
        //用于创建线程的，构造方法传入的 前面刚分析
        final ThreadFactory threadFactory;
        //原子性引用
        final AtomicReference<FixedSchedulerPool> pool;
        public EventLoopsScheduler(ThreadFactory threadFactory) {
    
            this.threadFactory = threadFactory;
            //放入一个默认的NONE对象 后面在start方法后为替换一个新的
            this.pool = new AtomicReference<FixedSchedulerPool>(NONE);
            start();
        }
    
        @Override
        public void start() {
            //
            FixedSchedulerPool update = new FixedSchedulerPool(threadFactory, MAX_THREADS);
            //原子性引用替换新对象
            if (!pool.compareAndSet(NONE, update)) {
                update.shutdown();
            }
        }

到这里分析完了一个**Scheduler**对象创建流程。  
做一个小总结：

subscribeOn(Schedulers.io());

`Schedulers.io()`方法大致流程，首先创建`Schedulers`单例对象（如果存在不创建），在创建`Schedulers`构造方法中完成自带的一些`Scheduler`对象创建（如`IO`和`computation`）

**subscribeOn（）** 方法分析

//Observable.java
     public final Observable<T> subscribeOn(Scheduler scheduler) {
            return subscribeOn(scheduler, !(this.onSubscribe instanceof OnSubscribeCreate));
        }

// Observable.java
    public final Observable<T> subscribeOn(Scheduler scheduler, boolean requestOn) {        //false
            if (this instanceof ScalarSynchronousObservable) {
                return ((ScalarSynchronousObservable<T>)this).scalarScheduleOn(scheduler);
            }
    
            //requestOn为true
            //OperatorSubscribeOn 这个类主要是创建Subcribe类（注：请区分OnSubscribe和Subcriber）
            //（Obserable创建需要Subcrib对象作为参数，此对象就只一个方法call（）方法）
            //unsafeCreate方法主要是创建Observerable对象
            return unsafeCreate(new OperatorSubscribeOn<T>(this, scheduler, requestOn));
        }

//Observable.java
      public static <T> Observable<T> unsafeCreate(OnSubscribe<T> f) {
                 //RxJavaHooks.onCreate忽略
                 //理解为new Observable<T>(f);即可
                return new Observable<T>(RxJavaHooks.onCreate(f));
        }

从上面的分析可以得知`subscribeOn`和 map等操作符一样会创建一个`Observable`对象返回，在最后调用`Observable`的订阅方法`subscribe`的时候`Observable`中传入的`OnSubscribe`的 `call`方法中完成重要的逻辑，因为这个分析网上实在太多了，我这里直接总结即可。（[不理解可以参考此文章，不然这篇文章会脱离讨论中心][Link 1]）

从上面的分析中`subscribeOn`方法创建了一个`OperatorSubscribeOn`的`OnSubscribe`对象

回顾下

unsafeCreate(new OperatorSubscribeOn<T>(this, scheduler, requestOn))

所以来看看把

// OperatorSubscribeOn.java
    public final class OperatorSubscribeOn<T> implements OnSubscribe<T> {
        
    
        final Scheduler scheduler;
        final Observable<T> source;
        final boolean requestOn;
    
        public OperatorSubscribeOn(Observable<T> source, Scheduler scheduler, boolean requestOn) {
            //= =嘻嘻
            this.scheduler = scheduler;
             //原来的被观测者
            this.source = source;
            //传入true
            this.requestOn = requestOn;
        }
        //订阅的时候调用此方法
        //所以我们可以从这个方法 查看如果利用scheduler来进行线程切换的
        @Override
        public void call(final Subscriber<? super T> subscriber) {
            final Worker inner = scheduler.createWorker();
    
            SubscribeOnSubscriber<T> parent = new SubscribeOnSubscriber<T>(subscriber, requestOn, inner, source);
            subscriber.add(parent);
            subscriber.add(inner);
    
            inner.schedule(parent);
        }
    
        static final class SubscribeOnSubscriber<T> extends Subscriber<T> implements Action0 {
    
            final Subscriber<? super T> actual;
    
            final boolean requestOn;
    
            final Worker worker;
    
            Observable<T> source;
    
            Thread t;
    
            SubscribeOnSubscriber(Subscriber<? super T> actual, boolean requestOn, Worker worker, Observable<T> source) {
                this.actual = actual;
                this.requestOn = requestOn;
                this.worker = worker;
                this.source = source;
            }
    
            @Override
            public void onNext(T t) {
                actual.onNext(t);
            }
    
            @Override
            public void onError(Throwable e) {
                try {
                    actual.onError(e);
                } finally {
                    worker.unsubscribe();
                }
            }
    
            @Override
            public void onCompleted() {
                try {
                    actual.onCompleted();
                } finally {
                    worker.unsubscribe();
                }
            }
    
            @Override
            public void call() {
                Observable<T> src = source;
                source = null;
                t = Thread.currentThread();
                src.unsafeSubscribe(this);
            }
    
            @Override
            public void setProducer(final Producer p) {
                actual.setProducer(new Producer() {
                    @Override
                    public void request(final long n) {
                        if (t == Thread.currentThread() || !requestOn) {
                            p.request(n);
                        } else {
                            worker.schedule(new Action0() {
                                @Override
                                public void call() {
                                    p.request(n);
                                }
                            });
                        }
                    }
                });
            }
        }
    }

因为在被订阅的时候此方法的`call`方法将会被会回调,所以我们细看此方法

//OperatorSubscribeOn.java
     @Override
        public void call(final Subscriber<? super T> subscriber) {
            //内部就是一个SchedulerThreadPoolExecutor
            final Worker inner = scheduler.createWorker();
    
            SubscribeOnSubscriber<T> parent = new SubscribeOnSubscriber<T>(subscriber, requestOn, inner, source);
    
            subscriber.add(parent);
            subscriber.add(inner);
    
            inner.schedule(parent);
        }

`scheduler.createWorker();` 我们看看源码

**Scheduler**对象实例为

//EventLoopsScheduler.java
        @Override
        public Worker createWorker() {
            return new EventLoopWorker(pool.get().getEventLoop());
        }

pool对象声明(如果忘记回头看看**FixedSchedulerPool**这个类,这个类内部有线程池数组)

//EventLoopsScheduler.java
    
       final AtomicReference<FixedSchedulerPool> pool;

`pool.get()`方法返回原子性引用的对象

//FixedSchedulerPool.java
    /*
    获取其对象数组中的一个实例化对象.
    该对象包含一个线程池对象实例(PoolWorker)
    */
     public PoolWorker getEventLoop() {
                int c = cores;
                if (c == 0) {
                    return SHUTDOWN_WORKER;
                }
                // simple round robin, improvements to come
                return eventLoops[(int)(n++ % c)];
            }

那么继续回到前面看的话可以编写以下伪代码

//EventLoopsScheduler.java
        @Override
        public Worker createWorker() {
            return new EventLoopWorker(线程池对象实例);
        }

`EventLoopWorker`对象看看

static final class EventLoopWorker extends Scheduler.Worker {
            private final SubscriptionList serial = new SubscriptionList();
            private final CompositeSubscription timed = new CompositeSubscription();
            private final SubscriptionList both = new SubscriptionList(serial, timed);
            private final PoolWorker poolWorker;
    
            EventLoopWorker(PoolWorker poolWorker) {
                this.poolWorker = poolWorker;
    
            }
    
            @Override
            public void unsubscribe() {
                both.unsubscribe();
            }
    
            @Override
            public boolean isUnsubscribed() {
                return both.isUnsubscribed();
            }
    
    
            @Override
            public Subscription schedule(final Action0 action) {
                if (isUnsubscribed()) {
                    return Subscriptions.unsubscribed();
                }
    
                return poolWorker.scheduleActual(new Action0() {
                    @Override
                    public void call() {
                        if (isUnsubscribed()) {
                            return;
                        }
                        action.call();
                    }
                }, 0, null, serial);
            }
            /*
            最终调用到这个函数
            */
            @Override
            public Subscription schedule(final Action0 action, long delayTime, TimeUnit unit) {
                if (isUnsubscribed()) {
                    return Subscriptions.unsubscribed();
                }
                /*
                    提交到线程池中运行,由于比较容易这里就不点进入看了
                */
                return poolWorker.scheduleActual(new Action0() {
                    @Override
                    public void call() {
                        if (isUnsubscribed()) {
                            return;
                        }
                        action.call();
                    }
                }, delayTime, unit, timed);
            }
        }

再回头看看我们前面的代码

@Override
        public void call(final Subscriber<? super T> subscriber) {
            final Worker inner = scheduler.createWorker();
    
            SubscribeOnSubscriber<T> parent = new SubscribeOnSubscriber<T>(subscriber, requestOn, inner, source);
            subscriber.add(parent);
            subscriber.add(inner);
            //这行执行后把会在线程池中调用parent的call方法
            inner.schedule(parent);
        }

`SubscribeOnSubscriber`的`call`方法

//SubscribeOnSubscriber.java
            @Override
            public void call() {
                //原来的被观察对象
                Observable<T> src = source;
                source = null;
    
                t = Thread.currentThread();
                //用自身去订阅,那么会触发自身的setProducer方法 从而从触发自身next方法
                //那么你在自身的next方法在调用 原本的订阅者的回调
    
                src.unsafeSubscribe(this);
            }

所以总的来看这个类

//OperatorSubscribeOn.java
    public final class OperatorSubscribeOn<T> implements OnSubscribe<T> {
        
    
        final Scheduler scheduler;
        final Observable<T> source;
        final boolean requestOn;
    
        public OperatorSubscribeOn(Observable<T> source, Scheduler scheduler, boolean requestOn) {
            this.scheduler = scheduler;
            this.source = source;
            this.requestOn = requestOn;
        }
    
        @Override
        public void call(final Subscriber<? super T> subscriber) {
            final Worker inner = scheduler.createWorker();
    
            SubscribeOnSubscriber<T> parent = new SubscribeOnSubscriber<T>(subscriber, requestOn, inner, source);
            subscriber.add(parent);
            subscriber.add(inner);
            //触发SubscribeOnSubscriber的call方法(在线程池中调用)
            inner.schedule(parent);
    
        }
        /*
            用来订阅原本的事件
        */
        static final class SubscribeOnSubscriber<T> extends Subscriber<T> implements Action0 {
    
            final Subscriber<? super T> actual;
    
            final boolean requestOn;
    
            final Worker worker;
    
            Observable<T> source;
    
            Thread t;
    
            SubscribeOnSubscriber(Subscriber<? super T> actual, boolean requestOn, Worker worker, Observable<T> source) {
                this.actual = actual;
                this.requestOn = requestOn;
                this.worker = worker;
                this.source = source;
            }
    
            @Override
            public void onNext(T t) {
                actual.onNext(t);
            }
    
            @Override
            public void onError(Throwable e) {
                try {
                    actual.onError(e);
                } finally {
                    worker.unsubscribe();
                }
            }
    
            @Override
            public void onCompleted() {
                try {
                    actual.onCompleted();
                } finally {
                    worker.unsubscribe();
                }
            }
    
            @Override
            public void call() {
                Observable<T> src = source;
                source = null;
                t = Thread.currentThread();
                //订阅的原本的被观察者.会调用 SubscribeOnSubscriber的setProducer
                //续而调用Producer.request
                src.unsafeSubscribe(this);
            }
    
            /*
             这里会调用Producer.request 然后回调自身的onNext等方法 
             Producer主要用于处理背压 .= =为了不脱离主题 直接一笔而过
            */
            @Override
            public void setProducer(final Producer p) {
                actual.setProducer(new Producer() {
                    @Override
                    public void request(final long n) {
                        if (t == Thread.currentThread() || !requestOn) {
                            p.request(n);
                        } else {
                            worker.schedule(new Action0() {
                                @Override
                                public void call() {
                                    p.request(n);
                                }
                            });
                        }
                    }
                });
            }
        }
    }

= =其实你多读几篇你就会明白**subscribeOn**仅有第一次生效.因为最原始的观察者的回调方法是第一个新的**Observerable**包裹,在这个新的包裹中的线程池中运行.假设第二次调用**subscribeOn** 只会包裹第一个新的`Observerable`(第一个新生成的`Observerable`代码会在第二个`Observerable`中的线程池运行,而第一个`Observerable`重新再开一个线程池再去运行原始的`Observerable`)

举个例子:

new Thread() {
    
                @Override//假设这是第二个Observerable
                public void run() {
                    super.run();
    
                    new Thread() {
                        @Override//假设这是第一个Observerable
                        public void run() {
                            super.run();
                /**
                 * doSomeThing()
                 * 这里运行最原始的Observerable代码
                 *那么这里将会运行在最近包裹的Thread线程中
                 *也就是第一个假设这是第一个Observerable
                 *
                 */
    
    
                        }
                    };
                }
            };

[Link 1]: https://blog.piasy.com/2016/09/15/Understand-RxJava/