Rxjava 提供了subscribeOn()方法来用于每个observable对象的操作符在哪个线程上运行

Rxjava 提供了ObserveOn()方法来用于每个Subscriber(Observer)对象的操作符在哪个线程上运行

线程切换的时候subscribeOn()只被执行一次。如果出现多次，那么以第一次出现是用的那个线程为准。 ObserverOnobserveOn()改变调用它之后代码的线程

============================================

为什么多次调用subscribeOn()却只有第一个起作用？
为什么多次调用observeOn()却可以切换到不同线程
observeOn()后能不能再次调用subscribeOn()？

如果你有这些疑问，那接下来的内容必定能解决你心头的疑惑

subscribeOn()和observeOn()的区别

subscribeOn()和observeOn()都是用来切换线程用的

subscribeOn()改变调用它之前代码的线程
observeOn()改变调用它之后代码的线程

这里给出下面示例中用到的两个函数

//用指定的名称新建一个线程
public static Scheduler getNamedScheduler(final String name) {return Schedulers.from(Executors.newCachedThreadPool(new ThreadFactory() {@Overridepublic Thread newThread(@android.support.annotation.NonNull Runnable runnable) {return new Thread(runnable, name);}}));}//打印当前线程的名称
public static void threadInfo(String caller) {System.out.println(caller + " => " + Thread.currentThread().getName());}

一、subscribeOn()

在讲解他的原理之前，先来一个简单的例子，有个感性认识，学起来更轻松

先说结论：subscribeOn 作用于该操作符之前的 Observable 的创建操符作以及 doOnSubscribe 操作符，换句话说就是 doOnSubscribe 以及 Observable 的创建操作符总是被其之后最近的 subscribeOn 控制。没看懂不要紧，看下面代码和图你就懂了。

Observable.create(new Observable.OnSubscribe<String>() {@Overridepublic void call(Subscriber<? super String> subscriber) {threadInfo("OnSubscribe.call()");subscriber.onNext("RxJava");}}).subscribeOn(getNamedScheduler("create之后的subscribeOn")).doOnSubscribe(() -> threadInfo(".doOnSubscribe()-1")).subscribeOn(getNamedScheduler("doOnSubscribe1之后的subscribeOn")).doOnSubscribe(() -> threadInfo(".doOnSubscribe()-2")).subscribe(s -> {threadInfo(".onNext()");System.out.println(s + "-onNext");});

结果如下：

.doOnSubscribe()-2 => main
.doOnSubscribe()-1 => doOnSubscribe1之后的subscribeOn
OnSubscribe.call() => create之后的subscribeOn
.onNext() => create之后的subscribeOn
RxJava-onNext

3号框中的.doOnSubscribe(() -> threadInfo(“.doOnSubscribe()-2”)) 的之后由于没有subscribeOn操作符所以回调到该段代码被调用的线程（即主线程）

由于 subscribe 之前没有使用observeOn 指定Scheduler，所以.onNext()的线程是和OnSubscribe.call()使用相同的Scheduler 。

下面通过源码来分析一下：

1、示例代码：

 Observable.create(new Observable.OnSubscribe<String>() {@Overridepublic void call(Subscriber<? super String> subscriber) {subscriber.onNext("a");subscriber.onNext("b");subscriber.onCompleted();}}).subscribeOn(Schedulers.io()).subscribe(new Observer<String>() {@Overridepublic void onCompleted() {}@Overridepublic void onError(Throwable e) {}@Overridepublic void onNext(String integer) {System.out.println(integer);}});

运行如下：

a
b

2、subscribeOn()源代码

public final Observable<T> subscribeOn(Scheduler scheduler) {if (this instanceof ScalarSynchronousObservable) {return ((ScalarSynchronousObservable<T>)this).scalarScheduleOn(scheduler);}return create(new OperatorSubscribeOn<T>(this, scheduler));}

很明显，会走if之外的方法。

在这里我们可以看到，又创建了一个OperatorSubscribeOn对象，但创建时传入的参数为OperatorSubscribeOn(this,scheduler)，我们看一下此对象以及其对应的构造方法

3、create()的源代码：

public static <T> Observable<T> create(OnSubscribe<T> f) {return new Observable<T>(hook.onCreate(f));}

我们看到这个方法，使用OperatorSubscribeOn这个类，来创建一个新的Observable，那就把它叫做Observable_2，把原来的Observable叫做Observable_1

4、OperatorSubscribeOn类的源代码：

public final class OperatorSubscribeOn<T> implements OnSubscribe<T> {final Scheduler scheduler;final Observable<T> source;public OperatorSubscribeOn(Observable<T> source, Scheduler scheduler) {this.scheduler = scheduler;this.source = source;}@Overridepublic void call(final Subscriber<? super T> subscriber) {final Worker inner = scheduler.createWorker();subscriber.add(inner);inner.schedule(new Action0() {@Overridepublic void call() {final Thread t = Thread.currentThread();Subscriber<T> s = new Subscriber<T>(subscriber) {@Overridepublic void onNext(T t) {subscriber.onNext(t);}@Overridepublic void onError(Throwable e) {try {subscriber.onError(e);} finally {inner.unsubscribe();}}@Overridepublic void onCompleted() {try {subscriber.onCompleted();} finally {inner.unsubscribe();}}@Overridepublic void setProducer(final Producer p) {subscriber.setProducer(new Producer() {@Overridepublic void request(final long n) {if (t == Thread.currentThread()) {p.request(n);} else {inner.schedule(new Action0() {@Overridepublic void call() {p.request(n);}});}}});}};source.unsafeSubscribe(s);}});}
}

OperatorSubscribeOn类implements 了Onsubscribe接口，并实现call()方法
OperatorSubscribeOn的构造方法，
- 保存了Observable对象，就是调用了subscribeOn()方法的Observable对象
- 并保存了Scheduler对象。

这里做个总结。

把Observable.create()创建的称之为Observable_1,OnSubscribe_1。
把subscribeOn()创建的称之为Observable_2，OnSubscribe_2

Observable_1是由示例代码的第1、2行创建的
OperatorSubscribeOn类是implements Onsubscribe接口的，所以可以当做Onsubscribe类使用。（OnSubscribe_2）
并且OnSubscribe_2中保存了Observable_1的应用，即source。（在OperatorSubscribeOn源代码的第8行）
在subscribeOn()源代码的倒数第二行，create(new OperatorSubscribeOn<T>(this, scheduler))返回新创建的Observable_2对象。

4.1、分析call()方法。

inner.schedule()改变了线程，此时Action的call()运行在指定的线程中。
把示例代码中的Subscriber包装了一层，赋给对象S（Subscriber_2）。见上面代码21行。
source.unsafeSubscribe(s);，
- 注意：source是Observable_1对象，这里的s就是Subscriber_2
- 因为调用过subscribeOn(Schedulers.io())后，返回Observable_2对象，所以示例代码第13行代码的subscribe()就是Observable_2.subscribe()，也就是执行OnSubscribe_2的call()方法(即OperatorSubscribeOn类的源代码的第12行)。

4.2 看一下source.unsafeSubscribe(s);（第65行）代码都做了什么

这里的source就是Observable_1，s是Subscriber_2

unsafeSubscribe()源代码：

public final Subscription unsafeSubscribe(Subscriber<? super T> subscriber) {try {// new Subscriber so onStart itsubscriber.onStart();// allow the hook to intercept and/or decoratehook.onSubscribeStart(this, onSubscribe).call(subscriber);return hook.onSubscribeReturn(subscriber);} catch (Throwable e) {// special handling for certain Throwable/Error/Exception typesExceptions.throwIfFatal(e);// if an unhandled error occurs executing the onSubscribe we will propagate ittry {subscriber.onError(hook.onSubscribeError(e));} catch (Throwable e2) {Exceptions.throwIfFatal(e2);// if this happens it means the onError itself failed (perhaps an invalid function implementation)// so we are unable to propagate the error correctly and will just throwRuntimeException r = new RuntimeException("Error occurred attempting to subscribe [" + e.getMessage() + "] and then again while trying to pass to onError.", e2);// TODO could the hook be the cause of the error in the on error handling.hook.onSubscribeError(r);// TODO why aren't we throwing the hook's return value.throw r;}return Subscriptions.unsubscribed();}}

关键代码：

hook.onSubscribeStart(this, onSubscribe).call(subscriber);

该方法即调用了OnSubscribe_1.call()方法。

注意，此时的call()方法在我们指定的线程中运行。起到了改变线程的作用。

对于以上线程，我们可以总结，其有如下流程：

Observable.create() : 创建了Observable_1和OnSubscribe_1;
subscribeOn()：创建Observable_2和OperatorSubscribeOn(OnSubscribe_2)，同时OperatorSubscribeOn保存了Observable_1的引用。
示例代码中的subscribe(Observer) 实际上就是调用Observable_2.subscribe(Observer):
- 调用OperatorSubscribeOn的call()。call()改变了线程的运行，并且调用了Observable_1.unsafeSubscribe(s);
- Observable_1.unsafeSubscribe(s);，该方法的实现中调用了OnSubscribe_1的call()。

这样就实现了在指定线程运行OnSubscribe的call()函数，无论我们的subscribeOn()放在哪里,他改变的是subscribe()的过程，而不是onNext()的过程。

那么如果有多个subscribeOn()，那么线程会怎样执行呢。如果按照我们的逻辑，有以下程序

Observable.just("ss") .subscribeOn(Schedulers.io())   // ----1---.subscribeOn(Schedulers.newThread()) //----2----.subscribe(new Action1<String>() {@Overridepublic void call(String s) {}});

那么，我们根据之前的源码分析其执行逻辑。

Observable.just(“ss”),创建Observable，OnSubscribe
Observable_1.subscribeOn(Schedulers.io()):创建Observable_1,OperatorSubscribeOn_1并保存Observable的引用。
Observable_2.subscribeOn(Schedulers.newThread())：创建Observable_2,OperatorSubscribeOn_2并保存Observable_1的引用。
Observable_3.subscribe():
- 调用OperatorSubscribeOn_2.call(),改变线程为Schedulers.newThread()。
- 调用OperatorSubscribeOn_1.call(),改变线程为Schedulers.io()。
- 调用OnSubscribe.call(),此时call()运行在Schedulers.io()。

根据以上逻辑分析，会按照1的线程进行执行。

二、observeOn()

先说结论：observeOn作用于该操作符之后操作符直到出现新的observeOn操作符

举个例子：

Observable.just("RxJava").observeOn(getNamedScheduler("map之前的observeOn")).map(s -> {threadInfo(".map()-1");return s + "-map1";}).map( s -> {threadInfo(".map()-2");return s + "-map2";}).observeOn(getNamedScheduler("subscribe之前的observeOn")).subscribe(s -> {threadInfo(".onNext()");System.out.println(s + "-onNext");});

结果如下：

.map()-1 => map之前的observeOn
.map()-2 => map之前的observeOn
.onNext() => subscribe之前的observeOn
RxJava-map1-map2-onNext

下面通过源码来进行分析：

1、observeOn()源码

public final Observable<T> observeOn(Scheduler scheduler) {return observeOn(scheduler, RxRingBuffer.SIZE);}public final Observable<T> observeOn(Scheduler scheduler, int bufferSize) {return observeOn(scheduler, false, bufferSize);}public final Observable<T> observeOn(Scheduler scheduler, boolean delayError, int bufferSize) {if (this instanceof ScalarSynchronousObservable) {return ((ScalarSynchronousObservable<T>)this).scalarScheduleOn(scheduler);}return lift(new OperatorObserveOn<T>(scheduler, delayError, bufferSize));}

这里引出了lift()函数

public final <R> Observable<R> lift(final Operator<? extends R, ? super T> operator) {return new Observable<R>(new OnSubscribeLift<T, R>(onSubscribe, operator));}

关于lift的详细介绍，如果不明白lift的原理，参考这里：RxJava 之二—— Lift()详解

用OperatorObserveOn对象，创建OnSubscribeLift对象（实现了OnSubscribe接口），接着创建Observable对象。为了加以区分，这里我们把OnSubscribeLift叫做OnSubscribe_2，Observable叫做Observable_2。

2、OperatorObserveOn代码：

public final class OperatorObserveOn<T> implements Operator<T, T> {private final Scheduler scheduler;private final boolean delayError;private final int bufferSize;/*** @param scheduler the scheduler to use* @param delayError delay errors until all normal events are emitted in the other thread?*/public OperatorObserveOn(Scheduler scheduler, boolean delayError) {this(scheduler, delayError, RxRingBuffer.SIZE);}/*** @param scheduler the scheduler to use* @param delayError delay errors until all normal events are emitted in the other thread?* @param bufferSize for the buffer feeding the Scheduler workers, defaults to {@code RxRingBuffer.MAX} if <= 0*/public OperatorObserveOn(Scheduler scheduler, boolean delayError, int bufferSize) {this.scheduler = scheduler;this.delayError = delayError;this.bufferSize = (bufferSize > 0) ? bufferSize : RxRingBuffer.SIZE;}@Overridepublic Subscriber<? super T> call(Subscriber<? super T> child) {if (scheduler instanceof ImmediateScheduler) {// avoid overhead, execute directlyreturn child;} else if (scheduler instanceof TrampolineScheduler) {// avoid overhead, execute directlyreturn child;} else {ObserveOnSubscriber<T> parent = new ObserveOnSubscriber<T>(scheduler, child, delayError, bufferSize);parent.init();return parent;}}public static <T> Operator<T, T> rebatch(final int n) {return new Operator<T, T>() {@Overridepublic Subscriber<? super T> call(Subscriber<? super T> child) {ObserveOnSubscriber<T> parent = new ObserveOnSubscriber<T>(Schedulers.immediate(), child, false, n);parent.init();return parent;}};}/** Observe through individual queue per observer. */static final class ObserveOnSubscriber<T> extends Subscriber<T> implements Action0 {final Subscriber<? super T> child;final Scheduler.Worker recursiveScheduler;final NotificationLite<T> on;final boolean delayError;final Queue<Object> queue;/** The emission threshold that should trigger a replenishing request. */final int limit;// the status of the current streamvolatile boolean finished;final AtomicLong requested = new AtomicLong();final AtomicLong counter = new AtomicLong();/** * The single exception if not null, should be written before setting finished (release) and read after* reading finished (acquire).*/Throwable error;/** Remembers how many elements have been emitted before the requests run out. */long emitted;// do NOT pass the Subscriber through to couple the subscription chain ... unsubscribing on the parent should// not prevent anything downstream from consuming, which will happen if the Subscription is chainedpublic ObserveOnSubscriber(Scheduler scheduler, Subscriber<? super T> child, boolean delayError, int bufferSize) {this.child = child;this.recursiveScheduler = scheduler.createWorker();this.delayError = delayError;this.on = NotificationLite.instance();int calculatedSize = (bufferSize > 0) ? bufferSize : RxRingBuffer.SIZE;// this formula calculates the 75% of the bufferSize, rounded up to the next integerthis.limit = calculatedSize - (calculatedSize >> 2);if (UnsafeAccess.isUnsafeAvailable()) {queue = new SpscArrayQueue<Object>(calculatedSize);} else {queue = new SpscAtomicArrayQueue<Object>(calculatedSize);}// signal that this is an async operator capable of receiving this manyrequest(calculatedSize);}void init() {// don't want this code in the constructor because `this` can escape through the // setProducer callSubscriber<? super T> localChild = child;localChild.setProducer(new Producer() {@Overridepublic void request(long n) {if (n > 0L) {BackpressureUtils.getAndAddRequest(requested, n);schedule();}}});localChild.add(recursiveScheduler);localChild.add(this);}@Overridepublic void onNext(final T t) {if (isUnsubscribed() || finished) {return;}if (!queue.offer(on.next(t))) {onError(new MissingBackpressureException());return;}schedule();}@Overridepublic void onCompleted() {if (isUnsubscribed() || finished) {return;}finished = true;schedule();}@Overridepublic void onError(final Throwable e) {if (isUnsubscribed() || finished) {RxJavaHooks.onError(e);return;}error = e;finished = true;schedule();}protected void schedule() {if (counter.getAndIncrement() == 0) {recursiveScheduler.schedule(this);}}// only execute this from schedule()@Overridepublic void call() {long missed = 1L;long currentEmission = emitted;// these are accessed in a tight loop around atomics so// loading them into local variables avoids the mandatory re-reading// of the constant fieldsfinal Queue<Object> q = this.queue;final Subscriber<? super T> localChild = this.child;final NotificationLite<T> localOn = this.on;// requested and counter are not included to avoid JIT issues with register spilling// and their access is is amortized because they are part of the outer loop which runs// less frequently (usually after each bufferSize elements)for (;;) {long requestAmount = requested.get();while (requestAmount != currentEmission) {boolean done = finished;Object v = q.poll();boolean empty = v == null;if (checkTerminated(done, empty, localChild, q)) {return;}if (empty) {break;}localChild.onNext(localOn.getValue(v));currentEmission++;if (currentEmission == limit) {requestAmount = BackpressureUtils.produced(requested, currentEmission);request(currentEmission);currentEmission = 0L;}}if (requestAmount == currentEmission) {if (checkTerminated(finished, q.isEmpty(), localChild, q)) {return;}}emitted = currentEmission;missed = counter.addAndGet(-missed);if (missed == 0L) {break;}}}boolean checkTerminated(boolean done, boolean isEmpty, Subscriber<? super T> a, Queue<Object> q) {if (a.isUnsubscribed()) {q.clear();return true;}if (done) {if (delayError) {if (isEmpty) {Throwable e = error;try {if (e != null) {a.onError(e);} else {a.onCompleted();}} finally {recursiveScheduler.unsubscribe();}}} else {Throwable e = error;if (e != null) {q.clear();try {a.onError(e);} finally {recursiveScheduler.unsubscribe();}return true;} elseif (isEmpty) {try {a.onCompleted();} finally {recursiveScheduler.unsubscribe();}return true;}}}return false;}}
}

虽然代码很长，但是也就是三部分

构造函数，
实现Operator所继承的Func1中的call()函数
静态内部类ObserveOnSubscriber< T>

下面来逐一分析：

因为调用Observable.等函数而需要创建的称之为Observable_1,Subscriber_1。
因为调用observeOn()而创建的称之为Observable_2，Subscriber_2

2.1、创建OperatorObserveOn对象

上面这段代码，主要功能就是创建OperatorObserveOn对象

既然是Operator，那么它的职责就是把一个Subscriber转换成另外一个Subscriber，

2.2、OperatorObserveOn对象中的call()函数返回ObserveOnSubscriber对象（Subscriber_2）

我们来看下call函数都做了什么：

ObserveOnSubscriber是一个静态类（第53行），创建一个ObserveOnSubscriber类（继承Subscriber< T>(Subscriber_2)）（OperatorObserveOn代码第35行），在参数中传入Subscriber_1（即局部变量child）和scheduler(指定线程)等参数。
调用了observeOn()，在subscribe()中调用onSubscribe.call(subscriber);时，就会调用上面代码第27行的call()，结果被传入到ObserveOnSubscriber的onNext()（第118行）。（如果不明白，请看RxJava 之二—— Lift()详解）

public void onNext(final T t) {if (isUnsubscribed() || finished) {return;}if (!queue.offer(on.next(t))) {onError(new MissingBackpressureException());return;}schedule();
}

这里做了两件事，

把执行的结果缓存到一个队列里，这里的on对象，不是Subscriber_1。
调用schedule()启动传入的线程所创建的worker

2.3、schedule()代码：

protected void schedule() {if (counter.getAndIncrement() == 0) {recursiveScheduler.schedule(this);}
}

recursiveScheduler 就是之前我们传入的Scheduler，就是在observeOn()传入的指定线程，例如：AndroidScheluders.mainThread()

2.4、我们看下在scheduler()中调用的call()方法代码，call()方法只能由scheduler()去调用执行

@Override
public void call() {...final Subscriber<? super T> localChild = this.child;for (;;) {...boolean done = finished;Object v = q.poll();boolean empty = v == null;if (checkTerminated(done, empty, localChild, q)) {return;}if (empty) {break;}localChild.onNext(localOn.getValue(v));...}if (emitted != 0L) {request(emitted);}
}

OK，在Scheduler启动后，我们在Observable.subscribe(a)传入的a就是这里的localChild（即Subscriber_1，是在第35行代码传递进来的），我们看到，在call中终于调用了它的onNext方法，把真正的结果传了出去，此时是工作在observeOn()指定的线程。

那么总结起来的结论就是：

observeOn 对调用之前的序列默不关心，也不会要求之前的序列运行在指定的线程上
observeOn 对之前的序列产生的结果先缓存起来，然后再在指定的线程上，推送给最终的subscriber

下面给出两次调用observeOn()的示意图

复杂情况

我们经常多次使用subscribeOn()切换线程，那么以后是否可以组合observeOn()和subscribeOn()达到自由切换的目的呢？

subscribeOn()改变的是subscribe()这句调用所在的线程，大多数情况，产生内容和消费内容是在同一线程的，所以改变了产生内容所在的线程，就改变了消费内容所在的线程。

对subscribeOn()的调用是自下向上，所以连续多次调用subscribeOn()，结果会被最上面的subscribeOn()覆盖。（生成和消费都会被覆盖）
observeOn()之上有subscribeOn()调用
observeOn()的工作原理是把消费结果先缓存，再切换到新线程上让原始消费者消费，它和生产者是没有一点关系的，就算subscribeOn()调用了，也只是改变observeOn()这个消费者所在的线程，和OperatorObserveOn中存储的原始消费者一点关系都没有，它还是由observeOn()控制。
observeOn()之下有subscribeOn()调用
这也不会改变observeOn()所指定的消费线程，因为observeOn()是自上而下调用，对subscribeOn()的调用是自下向上，在observeOn()指定的线程会覆盖下面subscribeOn()指定线程来去消费

用一张图来解释当多个 subscribeOn() 和 observeOn() 混合使用时，线程调度是怎么发生的（由于图中对象较多，相对于上面的图对结构做了一些简化调整）：

参考：http://blog.csdn.net/jdsjlzx/article/details/51685769
http://blog.csdn.net/jdsjlzx/article/details/51686152
https://segmentfault.com/a/1190000004856071
https://gank.io/post/560e15be2dca930e00da1083
你真的会用RxJava么?RxJava线程变换之observeOn与subscribeOn

observeOn()与subscribeOn()的详解相关推荐

RxJava操作符在android中的使用场景详解（一）
转载请注明出处:http://www.wangxinarhat.com/2016/04/19/2016-04-19-rxjava-android-operate1/ 最近学习了RxJava在andro ...
RxJava flatMap操作符用法详解
RxJava系列文章目录导读: 一.RxJava create操作符的用法和源码分析二.RxJava map操作符用法详解三.RxJava flatMap操作符用法详解四.RxJava conc ...
android ------- 开发者的 RxJava 详解
在正文开始之前的最后,放上 GitHub 链接和引入依赖的 gradle 代码: Github: https://github.com/ReactiveX/RxJava https://githu ...
Android高级开发之【RxJava】详解（附项目源码）
文章大纲一.什么是RxJava 二.为什么要用RxJava 三.RxJava使用详解四.项目源码下载一.什么是RxJava Rx(Reactive Extensions)是一个库,用来处理事件和 ...
Carson带你学Android：图文详解RxJava背压策略
前言 Rxjava,由于其基于事件流的链式调用.逻辑简洁 & 使用简单的特点,深受各大 Android开发者的欢迎. 本文主要讲解的是RxJava中的背压控制策略,希望你们会喜欢. Cars ...
RxJava2.x的集成及用法详解
目录目录主页中文资料 Rxjava是什么用途配置添加依赖即可基本概念 1. 被观察者. Observable 2. 观察者. Observer 3. 订阅. Subscribe 4. 事 ...
RxJava图文详解，可以说是很全了。
这里我拿出来给 Android 开发者的 RxJava 详解中的例子假设有这样一个需求:界面上有一个自定义的视图 imageCollectorView ,它的作用是显示多张图片,并能使用 addIm ...
扔物线------给 Android 开发者的 RxJava 详解
本文转载自扔物线的文章:http://gank.io/post/560e15be2dca930e00da1083 给 Android 开发者的 RxJava 详解 <p>作者:<a ...
Android 开发者的 RxJava 详解 - 作者：扔物线
前言我从去年开始使用 RxJava ,到现在一年多了.今年加入了 Flipboard 后,看到 Flipboard 的 Android 项目也在使用 RxJava ,并且使用的场景越来越多 .而最近 ...

observeOn()与subscribeOn()的详解