Rxjava 2.x 源码系列 - 线程切换（上）

时间 2019-12-07

标签 rxjava 2.x 源码系列线程切换栏目 Java 繁體版

原文原文链接

Rxjava 2.x 源码系列 - 基础框架分析java

Rxjava 2.x 源码系列 - 线程切换（上）c#

Rxjava 2.x 源码系列 - 线程切换（下）bash

Rxjava 2.x 源码系列 - 变换操做符 Map（上）app

前言

在上一篇博客 Rxjava 源码系列 - 基础框架分析，咱们分析了 Rxjava 的基础框架。框架

Observable 和 Observer 经过 subscribe() 方法实现订阅关系，从而 Observable 能够在须要的时候发出事件来通知 Observer，而且回调 Observer 的相应的方法。ide

用一张简单的流程图描述以下：函数

Observable#subscribeOn

在 Android 中，咱们知道默认都是执行在主线程的，那么 Rxjava 是如何实现线程切换的。post

Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                emitter.onNext("1");
                emitter.onNext("2");
                emitter.onNext("3");
                emitter.onComplete();
            }
        })
        .subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e("TAG", "onSubscribe(): ");
            }

            @Override
            public void onNext(String s) {
                Log.e("TAG", "onNext(): " + s);
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {
                Log.e("TAG", "onComplete(): ");
            }
        });
复制代码

咱们先来看一下 subscribeOn 方法，能够看到ui

@CheckReturnValue
@SchedulerSupport(SchedulerSupport.CUSTOM)
public final Observable<T> subscribeOn(Scheduler scheduler) {
    // scheduler 判空
    ObjectHelper.requireNonNull(scheduler, "scheduler is null");
    // 用 ObservableSubscribeOn 将 scheduler 包装 起来
    return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this, scheduler));
}


复制代码

而咱们从上一篇博客中知道，当咱们调用 observable.subscibe(observable) 的时候，最终会调用到具体的 observable 的实例的 subscribActual 方法。而这里具体的 observable 的实例为 ObservableSubscribeOn。this

接下来，咱们来看一下 ObservableSubscribeOn 这个类，能够看到继承 AbstractObservableWithUpstream ，而 AbstractObservableWithUpstream 继承 Observable，实现 HasUpstreamObservableSource 这个接口。

public final class ObservableSubscribeOn<T> extends AbstractObservableWithUpstream<T, T> {
    final Scheduler scheduler;

    public ObservableSubscribeOn(ObservableSource<T> source, Scheduler scheduler) {
        super(source);
        this.scheduler = scheduler;
    }

    @Override
    public void subscribeActual(final Observer<? super T> s) {
        final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(s);

        s.onSubscribe(parent);

        parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
    }
    
    ---
}


abstract class AbstractObservableWithUpstream<T, U> extends Observable<U> implements HasUpstreamObservableSource<T> {

    /** The source consumable Observable. */
    protected final ObservableSource<T> source;

    /**
     * Constructs the ObservableSource with the given consumable.
     * @param source the consumable Observable
     */
    AbstractObservableWithUpstream(ObservableSource<T> source) {
        this.source = source;
    }

    @Override
    public final ObservableSource<T> source() {
        return source;
    }

}

public interface HasUpstreamObservableSource<T> {
    /**
     * Returns the upstream source of this Observable.
     * <p>Allows discovering the chain of observables.
     * @return the source ObservableSource
     */
    ObservableSource<T> source();
}
复制代码

observableSubscribeOn 的 subscribeActual 方法，跟 ObservableCreate 的 subscribeActual 的套路差很少，它也是 Observable 的一个子类。只不过比 ObservableCreate 多实现了一个接口HasUpstreamObservableSource，这个接口颇有意思，他的 source() 方法返回类型是 ObservableSource（还记得这个类的角色吗？）。也就是说 ObservableSubscribeOn 这个 Observable 是一个拥有上游的 Observable 。他有一个很是关键的属性 source，这个 source 就表明了他的上游。

接下来咱们一块儿来看一下 ObservableSubscribeOn 的具体实现

public final class ObservableSubscribeOn<T> extends AbstractObservableWithUpstream<T, T> {
    final Scheduler scheduler;

    public ObservableSubscribeOn(ObservableSource<T> source, Scheduler scheduler) {
        super(source);
        this.scheduler = scheduler;
    }

    @Override
    public void subscribeActual(final Observer<? super T> s) {
        final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(s);

        s.onSubscribe(parent);

        parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
    }
}
复制代码

首先先来看他的构造函数，有两个参数 source ，scheduler。

source 表明上游的引用，是 Observable 的一个实例
scheduler 能够经过 Schedulers.newThread() 或者 Schedulers.io() 建立相应的实例

这里咱们先大概了解一下 Scheduler 是个什么东东，Scheduler 里面封装了 Worker 和 DisposeTask，下面会详细讲到。

Schedulers.newThread()

@NonNull
public static Scheduler newThread() {
    return RxJavaPlugins.onNewThreadScheduler(NEW_THREAD);
}


NEW_THREAD = RxJavaPlugins.initNewThreadScheduler(new NewThreadTask());
static final class NewThreadTask implements Callable<Scheduler> {
    @Override
    public Scheduler call() throws Exception {
        return NewThreadHolder.DEFAULT;
    }
}
static final class NewThreadHolder {
    static final Scheduler DEFAULT = new NewThreadScheduler();
}


复制代码

public static Scheduler io() {
    return RxJavaPlugins.onIoScheduler(IO);
}

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

static final class IOTask implements Callable<Scheduler> {
    @Override
    public Scheduler call() throws Exception {
        return IoHolder.DEFAULT;
    }
}
static final class IoHolder {
    static final Scheduler DEFAULT = new IoScheduler();
}
static final class IoHolder {
    static final Scheduler DEFAULT = new IoScheduler();
}
复制代码

咱们再回到 ObservableSubscribeOn 的 subscribeActual 方法，在上一篇博客的时候已经讲解 Observable 和 Observer 之间是怎样实现订阅关系的，这里就再也不具体展开了。

接下来，咱们重点关注这一行代码

parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
 
复制代码

咱们先来看一下 SubscribeTask 这个类，他是 ObservableSubscribeOn 的一个非静态内部类，能够看到其实也比较简单，他实现了 Runnable 接口，而且持有 parent 引用。

final class SubscribeTask implements Runnable {
    private final SubscribeOnObserver<T> parent;

    SubscribeTask(SubscribeOnObserver<T> parent) {
        this.parent = parent;
    }

    @Override
    public void run() {
        source.subscribe(parent);
    }
}
复制代码

而后在 run 方法中，经过 source.subscribe(parent) 创建联系。于是，当咱们的 SubscribeTask 的 run 方法运行在哪一个线程，相应的 observer 的 subscribe 方法就运行在哪一个线程。

这里可能会有人有疑问，SubscribeTask 没有 source 属性，它是怎么访问到 ObservableSubscribeOn 的属性的。

咱们知道 java 中，非静态内部类默认持有外部类的引用，于是他能够正常访问外部类 ObservableSubscribeOn 的 source 属性。

接着，咱们再来看一下 scheduler.scheduleDirect 这个方法

@NonNull
public Disposable scheduleDirect(@NonNull Runnable run, long delay, @NonNull TimeUnit unit) {
    final Worker w = createWorker();

    // 判断 run 是否为 null
    final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);

    DisposeTask task = new DisposeTask(decoratedRun, w);

    w.schedule(task, delay, unit);

    return task;
}


复制代码

首先，建立一个 Worker w
第二步，DisposeTask 将 decoratedRun 包装起来
第三步：w 去调度 task

这里咱们以 NewThreadScheduler 为例，来看看这个 Worker 究竟是什么？

public Worker createWorker() {
    return new NewThreadWorker(threadFactory);
}



public class NewThreadWorker extends Scheduler.Worker implements Disposable {
    private final ScheduledExecutorService executor;

    volatile boolean disposed;

    public NewThreadWorker(ThreadFactory threadFactory) {
        executor = SchedulerPoolFactory.create(threadFactory);
    }
    
    --- 
}


public static ScheduledExecutorService create(ThreadFactory factory) {
    final ScheduledExecutorService exec = Executors.newScheduledThreadPool(1, factory);
    if (PURGE_ENABLED && exec instanceof ScheduledThreadPoolExecutor) {
        ScheduledThreadPoolExecutor e = (ScheduledThreadPoolExecutor) exec;
        POOLS.put(e, exec);
    }
    return exec;
}



复制代码

从上面能够看到，其实 worker 里面封装了 executor（线程池），看到这里，相信你也基本明白 Rxjava 线程切换的原理了，其实很简单。

在 ObservableSubscribeOn subscribeActual 方法中， SubscribeTask 包装 parent（SubscribeOnObserver ，包装了 Observer），SubscribeTask 实现了 Runnable 接口，在 run 方法里面调用了 source.subscribe(parent)，于是 run 方法所执行的线程将由 worker 决定。这就是下游决定上游 observable 执行线程的原理。

接下来咱们再来看一下：DisposeTask

static final class DisposeTask implements Disposable, Runnable, SchedulerRunnableIntrospection {
        final Runnable decoratedRun;
        final Worker w;

        Thread runner;

        DisposeTask(Runnable decoratedRun, Worker w) {
            this.decoratedRun = decoratedRun;
            this.w = w;
        }

        @Override
        public void run() {
            runner = Thread.currentThread();
            try {
                decoratedRun.run();
            } finally {
                dispose();
                runner = null;
            }
        }

        @Override
        public void dispose() {
            if (runner == Thread.currentThread() && w instanceof NewThreadWorker) {
                ((NewThreadWorker)w).shutdown();
            } else {
                w.dispose();
            }
        }

        @Override
        public boolean isDisposed() {
            return w.isDisposed();
        }

        @Override
        public Runnable getWrappedRunnable() {
            return this.decoratedRun;
        }
    }
}
复制代码

// 将 新的 Disposable 设置给 parent ，方便取消订阅关系，
//（由于咱们对  Observer 进行相应的包装，原来的 parent 的 Disposable 已经不能表明最新的 Disposable）
parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
复制代码

DisposeTask 实现了 Disposable，Runnable ，SchedulerRunnableIntrospection 接口，Disposable 接口主要是用来取消订阅关系的 Disposable。

Observable#subscribeOn(Scheduler) 第一次有效原理

Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                Log.i(TAG, "subscribe: getName=" +Thread.currentThread().getName());
                emitter.onNext("1");
                emitter.onNext("2");
                emitter.onNext("3");
                emitter.onComplete();
            }
        }) // 进行两次 subscribeOn
        .subscribeOn(Schedulers.io()).subscribeOn(Schedulers.computation()).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e("TAG", "onSubscribe(): ");
            }

            @Override
            public void onNext(String s) {
                Log.e("TAG", "onNext(): " + s);
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {
                Log.e("TAG", "onComplete(): ");
            }
        });
复制代码

subscribe: getName=RxCachedThreadScheduler-1

若是将上述的 subscribeOn 的顺序置换

subscribeOn(Schedulers.computation()).subscribeOn(Schedulers.io())
复制代码

那么将打印出

subscribe: getName=RxComputationThreadPool-1

为何是第一次 Observable#subscribeOn(Scheduler) 才有效呢？

前面咱们分析到，Observable#subscribeOn(Scheduler) 其实是将 Observable#subscribe(Observer) 的操做放在了指定线程，当咱们调用 subcribe 的时候，它的过程是从下往上的，即下面的 Observable 调用上面的 Observanle。

因此对于咱们上面的第一个例子，他的调用流程是这样的：第三个 Observable 调用 Observable#subscribe(Observer) 启动订阅，在其内部会激活第二个 Observable 的 Observable#subscribe(Observer) 方法，可是此时该方法外部被套入了一个 Schedulers.computation() 线程

因而这个订阅的过程就被运行在了该线程中。用伪代码演示以下

public class Observable {
    // 第「二」个 Observable
    Observable source;
    Observer observer;

    public Observable(Observable source, Observer observer) {
        this.source = source;
        this.observer = observer;
    }

    public void subscribe(Observer Observer) {
        new Thread("computation") {
            @Override
            public void run() {
                // 第「二」个 Observable 订阅
                source.subscribe(observer);
            }
        }
    }
}
复制代码

再往上走，第二个 Observable 订阅内部会激活第一个 Observable 的 Observable#subscribe(Observer) 方法，一样的，该方法被套在了 Schedulers.io() 线程中，用伪代码演示

public class Observable {
    // 第「一」个 Observable
    Observable source;
    Observer observer;

    public Observable(Observable source, Observer observer) {
        this.source = source;
        this.observer = observer;
    }

    public void subscribe(Observer Observer) {
        new Thread("io") {
            @Override
            public void run() {
                // 第「一」个 Observable 订阅
                source.subscribe(observer);
            }
        }
    }
}
复制代码

此时到达第一个 Observable 了以后就要开始发射事件了，此时的执行线程很明显是 io 线程。还能够换成 Thread 伪代码来表示。

new Thread("computation") {
    @Override
    public void run() {
        // 第二个 Observable.subscribe(Observer) 的实质
        // 就是切换线程，效果相似以下
        new Thread("io") {
            @Override
            public void run() {
                // 第一个 Observable.subscribe(Observer) 的实质
                // 就是发射事件
                System.out.println("onNext(T)/onError(Throwable)/onComplete() 的执行线程是： " + Thread
                                   .currentThread().getName());
            }
        } .start();
    }
} .start();
复制代码

总结

用流程图描述以下：

参考博客：

友好 RxJava2.x 源码解析（二）线程切换

下一篇咱们将讲解到 observeOn(AndroidSchedulers.mainThread()) 的原理。

扫一扫，欢迎关注个人公众号。若是你有好的文章，也欢迎你的投稿。

Rxjava 2.x 源码系列 - 线程切换 （上）

前言

Observable#subscribeOn

Observable#subscribeOn(Scheduler) 第一次有效原理

总结

Rxjava 2.x 源码系列 - 线程切换（上）