RxJava 笔记 RxJava 其实就是一个实现异步操作 的库，由表示数据源的一组类、用于监听数据源的一组类，用于修改与合并数据的一组方法组成，使用的就是通用形式的观察者模式 。RxJava 优势在于能把复杂逻辑简洁化。
RxJava 的观察者模式
RxJava 有四个基本概念：
Observable（可观察者，即被观察者）
Observer （观察者）
subscribe（订阅）
事件
与传统的观察者模式不同，RxJava 的事件回调方法除了普通事件onNext()(相当于onClick()/onEvent())之外，还定义了两个特殊的事件回调onCompleted()和onError()。
onCompleted()：事件队列完结。RxJava 不仅把每个事件单独处理，还会把它们看做一个队列。RxJava 规定，当不会有新的onNext()发出时，需触发onCompleted()作为标志。onError()：事件队列异常。在事件处理过程中出现异常时，onError()会被触发，同时队列会自动终止，不允许再有事件发出。
基本实现 创建 Observer Observer 决定事件触发的时候将有怎样的行为。
1 2 3 4 5 6 7 8 9 10 Observer<String> observer = new Observer<String>(){ @Override public void onNext(String s){} @Overrider public void onCompleted(){} @Override public void onError(){} }
除了Observer接口之外，RxJava 还内置了一个实现了Observer的抽象类：Subscriber。Subscriber对Observer接口进行了一些扩展，但他们的基本使用方式是完全一样的：
1 2 3 4 5 6 7 8 9 10 Subscriber<String> subscriber = new Subscriber<String>(){ @Override public void onNext (String s) @Override public void onCompleted () @Override public void onError () }
不仅基本使用方式一样，实质上，在 RxJava 的 subscribe 过程中，Observer也总是先被转换成一个Subscriber再使用。所以如果只是使用基本功能，选择Observer和Subscriber是完全一样的。它们的区别对于使用者来说主要有两点：
onStart()：在 subscribe 刚开始，而事件还未发送之前被调用，可以用于做一些准备工作。需要注意的是，如果对准备工作的线程有要求（例如弹出一个显示进度的对话框，这必须在主线程执行），onStart()就不适用了，因为它总是在 subscribe 所发生的线程被调用，而不能指定线程。要指定的线程来做准备工作，可以使用doOnSubscribe()。unsubscribe()：这是Subscriber所实现的另一个接口Subscription的方法，用于取消订阅。在这个方法被调用后，Subscriber将不再接收事件。一般在这个方法调用前，可以使用isUnsubscribed()先判断下状态。unsubscribe()这个方法很重要，因为在subscribe()之后，Observable会持有Subscriber的引用，这个引用如若不能及时被释放，将有内存泄漏的风险。onPause()、onStop()等方法中）调用unsubscribe()来解除引用关系，以避免内存泄漏的发生。
创建 Observable Observable 即被观察者，它决定什么时候触发事件以及触发怎样的事件。
1 2 3 4 5 6 7 8 9 Observable observable = Observable.create(new Observable.OnSunscribe<String>(){ @Override public void call (Subscriber<? super String> subscriber) subscriber.onNext("Hello" ); subscriber.onNext("Hi" ); subscriber.onNext("Aloha" ); subscriber.onCompleted(); } });
当Observable被订阅的时候，OnSubscribe的call()方法会自动被调用，事件序列就会依照设定依次触发。create()方法是 RxJava 最基本创造事件序列的方法。还有一些方法：
1 Obervable observable = Observable.just("Hello" , "Hi" , "Aloha" );
1 2 String[] words = {"Hello" , "Hi" , "Aloha" }; Observable observable = Observable.from(words);
Subscribe（订阅） 1 2 3 observable.subscribe(observer); observable.subscribe(subscriber);
Observable.subscribe(Subscriber)内部实现（仅核心代码）：
1 2 3 4 5 public Subscription subscribe (Subscriber subscriber) subscriber.onStart(); onSubscribe.call(subscriber); return subscriber; }
从上可以看出，Observable并不是在创建的时候就立即开始发送事件，而是在它被订阅的时候才开始发送事件，即当subscribe()方法执行的时候
除了subscribe(Observer)和subscribe(Subscriber)，subscribe()还支持不完整定义的回调，RxJava 会自动根据定义创建出Subscriber。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 Action1<String> onNextAction = new Action1<String>(){ @Override public void call (String s) }; Action1<Throwable> onErrorAction = new Action1<Throwable>(){ @Override public void call (Throwable throwable) }; Action0 onCompletedAction = new Action0(){ @Override public void call () }; observable.subscribe(onNextAction); observable.subscribe(onNextAction, onErrorAction); observable.subscribe(onNextAction, onErrorAction, onCompletedAction);
简单使用：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 int drawableRes = ...;ImageView imageView = ...; Observable.create(new OnSubscribe<Drawable>() { @Override public void call (Subscriber<? super Drawable> subscriber) Drawable drawable = getTheme().getDrawable(drawableRes)); subscriber.onNext(drawable); subscriber.onCompleted(); } }).subscribe(new Observer<Drawable>() { @Override public void onNext (Drawable drawable) imageView.setImageDrawable(drawable); } @Override public void onCompleted () } @Override public void onError (Throwable e) Toast.makeText(activity, "Error!" , Toast.LENGTH_SHORT).show(); } });
线程控制—— Scheduler 在不指定线程的情况下，在哪个线程调用subscribe()，就在哪个线程生产事件，在哪个线程生产事件，就在哪个线程消费事件。如果需要切换线程，就需要用到Scheduler（调度器）。
Scheduler 的 API
Schedulers.immediate()：默认Schedulers.newThread()：总是启用新线程，并在新线程执行操作Schedulers.io()：I/O 操作（读写文件、读写数据库、网络信息交互等）。行为模式和newThread()差不多，区别在于io()的内部实现是用一个无数量上限的线程池，可以重用空闲的线程，因此多数情况下io()比newThread()更有效率。不要把计算工作放在io()中，可以避免创建不必要的线程。Schedulers.computation()计算所使用的，这个计算指的是 CPU 密集型计算，即不会被 I/O 等操作限制性能的计算，例如图形的计算。使用的固定的线程池，大小为 CPU 核数。AndroidSchedulers.mainThread()：指定的操作在Android 主线程运行
1 2 3 4 5 6 7 Observable.just(1 , 2 , 3 , 4 ) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(new Action1<Integer>(){ @Override public void call (Integer number) });
变换 RxJava 提供了对事件序列进行变换的支持。所谓变换，就是将事件队列中的对象或整个序列进行加工处理，转换成不同的事件或事件序列 。
API 1 2 3 4 5 6 7 8 9 10 11 12 Observable.just("images/logo.png" ) .map(new Func1<String, Bitmap>(){ @Override public Bitmap call (String filePath) return getBitmapFromPath(filePath); } }).subscribe(new Action1<Bitmap>(){ @Override public void call (Bitmap bitmap) showBitmap(bitmap); } });
Func1也是 RxJava 的一个接口。Func1和Action的区别在于，Func1包装的是有返回值的方法。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 Student[] students = ...; Subscriber<Course> subscriber = new Subscriber<Course>(){ @Override public void onNext (Course course) Log.d(tag, course.getName()); } ... }; Observable.from(students) .flatMap(new Func1<Student, Observable<Course>>(){ @Override public Observable<Course> call (Student student) return Observable.from(student.getCourses()); } }).subscriber(subscriber);
flatMap()和map()有一个相同点：把传入的参数转化后返回另一对象。flatMap()的原理步骤是：
使用传入的事件对象创建一个Observable对象
并不发送这个Observable，而是将它激活，于是它开始发送事件
每一个创建出来的Observable发送的事件，都被汇入同一个Observable，而这个Observable负责将这些事件同一交给Subscriber的回调方法。
变换的原理：lift() RxJava 提供了很多便捷的方法来实现事件序列的变换，这些变换虽然功能各有不同，但实质上都是针对事件序列的处理和再发送 。而在 RxJava 内部，它们是基于同一个基础的变换方法：lift(Operator)。
1 2 3 4 5 6 7 8 9 10 11 12 public <R> Observable<R> lift (Operator<? extends R, ? super T> operator) { return Observable.create(new OnSubscribe<R>(){ @Override public void call (Subscriber subscriber) Subscriber newSubscriber = opertor.call(subscriber); newSubscriber.onStart(); onSubscribe.call(newSubscriber); } }); }
在Onservable执行了lift(Operator)之后，会返回一个新的Observable，这个新的Observable会像一个代理一样，负责接收原始的Observable发出的事件，并在处理后发送给Subscriber。
两次和多次lift()同理
compose()：对 Observable 整体的变换list()是针对事件项和事件序列的，而compose()是针对Observable自身进行变换。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 public class LiftAllTransformer implements Observable .Transformer <Integer , String > @Override public Observable<String> call (Observable<Integer> observable) return observable .lift1() .lift2() .lift3() .lift4(); } } ... Transformer liftAll = new LiftAllTransformer(); observable1.compose(liftAll).subscribe(subscriber1); observable2.compose(liftAll).subscribe(subscriber2); observable3.compose(liftAll).subscribe(subscriber3); observable4.compose(liftAll).subscribe(subscriber4);
线程控制 PLUS 能不能多切换几次线程？—— 能！因为observeOn()指定的是Subscriber的线程，而这个Subscriber并不一定是subscribe()参数中的Subscriber，而是observeOn()执行时的当前Observable对应的Subscriber，即它的直接下级Subscriber。换句话说，observeOn()指定的是它之后的操作所在的线程。因此如果有多次切换线程的需求，只要在每个想要切换线程的位置调用一次observeOn()即可。
1 2 3 4 5 6 7 8 Observable.just(1 , 2 , 3 , 4 ) .subscribeOn(Schedulers.io()) .observeOn(Schedulers.newThread()) .map(mapOperator) .observeOn(Schedulers.io()) .map(maoOperator2) .observeOn(AndroidSchedulers.mainThread()) .subscribe(subscriber);
不过，不同于observeOn()的多次调用，subscribeOn()位置放在哪里都可以，但它是只能调用一次的。
Scheduler 的原理 subscribeOn()和observeOn()的内部实现，也是用的lift()。
subscribeOn()原理图：
observeOn()原理图：
从图中可以看出，subscribeOn()和observeOn()都做了线程切换的工作（图中的“schedule…”部位）。不同的是，subscribeOn()的线程切换发生在OnSubscribe中，即在它通知上一级OnSubscribe时，这时时间还没有开始发送subscribeOn()的线程控制可以从事件发出的开端就造成影响；而observeOn()的线程切换则发生在它内建的Subscriber中，即发生在它即将给下一级Subscriber发送事件时observeOn()控制的是它后面的线程。
图中共有 5 处含有对事件的操作。由图中可以看出，① 和 ② 两处受第一个subscribeOn()影响，运行在红色线程；③ 和 ④ 处受第一个observeOn()的影响，运行在绿色线程；⑤ 处受第二个observeOn()影响，运行在紫色线程；而第二个subscribeOn()，由于在通知过程中线程就被第一个subscribeOn()截断，因此对整个流程并没有任何影响。因此，当使用了多个subscribeOn()的时候，只有第一个subscribeOn()起作用。
doOnSubscribe() 与Subscriber.onStart()同样是在subscribe()调用后而且事件发送之前执行的，但区别是它可以指定线程。默认情况下，doOnSubscribe()执行在subscribe()发生的线程l；而如果在doOnSubscribe()之后有subscribeOn()的话 ，它将执行在离他最近的subscribeOn()所指定的线程。
1 2 3 4 5 6 7 8 9 10 Observable.create(onSubscribe) .subscrubeOn(Schedulers.io()) .doOnSubscribe(new Action0(){ @Override public void call () progressBar.setVisibility(View.VISIBLE); } }).subscribeOn(AndroidSchedulers.mainThread()) .observeOn(AndroidSchedulers.mainTread()) .subscribe(subscriber);
源码分析 基于 RxJava 2.1.2
Observable.create(oos)create()
1 2 3 4 public static <T> Observable<T> create (ObservableOnSubscribe<T> source) { ObjectHelper.requireNonNull(source, "source is null" ); return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source)); }
create() -> RxJavaPlugins.onAssembly()
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 @SuppressWarnings ({ "rawtypes" , "unchecked" }) @NonNull public static <T> Observable<T> onAssembly (@NonNull Observable<T> source) { Function<? super Observable, ? extends Observable> f = onObservableAssembly; if (f != null ) { return apply(f, source); } return source; }
钩子函数 ：函数主动找事件。
由上知，Observable.craete(oos)只是创建了一个 ObservableCreate 对象。
observeOn(AndroidSchedulers.mainThread())既然之前的create()方法创建了一个ObservableCreate对象并返回，也就是说observeOn(Schedulers.computation())这个方法调用的是ObservableCreate这个对象的方法。
1 public final class ObservableCreate <T > extends Observable <T >
没有看到observeOn()，看看父类Observable
1 2 3 public final Observable<T> observeOn (Scheduler scheduler) return observeOn(scheduler, false , bufferSize()); }
1 2 3 4 5 public final Observable<T> observeOn (Scheduler scheduler, boolean delayError, int bufferSize) ObjectHelper.requireNonNull(scheduler, "scheduler is null" ); ObjectHelper.verifyPositive(bufferSize, "bufferSize" ); return RxJavaPlugins.onAssembly(new ObservableObserveOn<T>(this , scheduler, delayError, bufferSize)); }
这里又是一个onAssembly()的调用
1 2 3 4 5 6 7 8 public static <T> Observable<T> onAssembly (@NonNull Observable<T> source) { Function<? super Observable, ? extends Observable> f = onObservableAssembly; if (f != null ) { return apply(f, source); } return source; }
同样的钩子方法，以后只需关心它传递的参数和返回值就行了。故observeOn()就是创建并返回一个ObservableObserveOn对象。
subscribeOn(Schedulers.computation())由于observeOn()创建并返回了一个ObservableObserveOn对象，所以这里调用的是ObservableObserveOn对象上的方法
1 2 3 public final class ObservableObserveOn <T > extends AbstractObservableWithUpstream <T , T > abstract class AbstractObservableWithUpstream <T , U > extends Observable <U > implements HasUpstreamObservableSource <T >
也是继承于Observable。
1 2 3 4 public final Observable<T> subscribeOn (Scheduler scheduler) ObjectHelper.requireNonNull(scheduler, "scheduler is null" ); return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this , scheduler)); }
但其实，observeOn()和subscribeOn()并没有什么使用顺序，而先前create()得到的Observable对象，我们也可以直接找到。subscribeOn()为 final 方法，创建了一个ObservableSubscribeOn对象并返回。
subscribe(o)Observable中的 final 方法：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 public final void subscribe (Observer<? super T> observer) ObjectHelper.requireNonNull(observer, "observer is null" ); try { observer = RxJavaPlugins.onSubscribe(this , observer); ObjectHelper.requireNonNull(observer, "Plugin returned null Observer" ); subscribeActual(observer); } catch (NullPointerException e) { throw e; } catch (Throwable e) { Exceptions.throwIfFatal(e); RxJavaPlugins.onError(e); NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS" ); npe.initCause(e); throw npe; } }
subscribe()
-> RxJavaPlugins.onSubscribe()
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 @SuppressWarnings ({ "rawtypes" , "unchecked" }) @NonNull public static <T> Observer<? super T> onSubscribe(@NonNull Observable<T> source, @NonNull Observer<? super T> observer) { BiFunction<? super Observable, ? super Observer, ? extends Observer> f = onObservableSubscribe; if (f != null ) { return apply(f, source, observer); } return observer; }
subscribe()
-> RxJavaPlugins.onSubscribe()
-> subscribeActual(observer)
1 protected abstract void subscribeActual (Observer<? super T> observer)
抽象方法，在ObservableSubscribeOn和ObservableObserceOn都有对应的重写实现 。我们先看ObservableSubscribeOn
1 2 3 4 5 6 @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))); }
这里需注意的是 s 这个参数，后面会有很多地方看到这个参数，一定要搞清楚这个参数是谁传递过来的。
比如说：
subscribe()
-> RxJavaPlugins.onSubscribe()
-> subscribeActual(observer)
-> scheduler.scheduleDirect()
scheduler变量就是我们使用subscribeOn()传递的参数：
1 2 3 4 public final Observable<T> subscribeOn (Scheduler scheduler) ObjectHelper.requireNonNull(scheduler, "scheduler is null" ); return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this , scheduler)); }
1 2 3 4 public ObservableSubscribeOn (ObservableSource<T> source, Scheduler scheduler) super (source); this .scheduler = scheduler; }
调用了schedulerDirect()
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 public Disposable scheduleDirect (@NonNull Runnable run) return scheduleDirect(run, 0L , TimeUnit.NANOSECONDS); } public Disposable scheduleDirect (@NonNull Runnable run, long delay, @NonNull TimeUnit unit) final Worker w = createWorker(); final Runnable decoratedRun = RxJavaPlugins.onSchedule(run); DisposeTask task = new DisposeTask(decoratedRun, w); w.schedule(task, delay, unit); return task; }
这两个方法是Scheduler，而当使用Schedulers.computation()时，返回的是一个ComputationScheduler对象，查看ComputationScheduler有没有重写这两个方法。
1 2 3 4 5 6 @NonNull @Override public Disposable scheduleDirect (@NonNull Runnable run, long delay, TimeUnit unit) PoolWorker w = pool.get().getEventLoop(); return w.scheduleDirect(run, delay, unit); }
w.scheduleDirect(run, delay, unit);
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 public Disposable scheduleDirect (final Runnable run, long delayTime, TimeUnit unit) ScheduledDirectTask task = new ScheduledDirectTask(RxJavaPlugins.onSchedule(run)); try { Future<?> f; if (delayTime <= 0L ) { f = executor.submit(task); } else { f = executor.schedule(task, delayTime, unit); } task.setFuture(f); return task; } catch (RejectedExecutionException ex) { RxJavaPlugins.onError(ex); return EmptyDisposable.INSTANCE; } }
线程池使用代码。其实就是向我们创建的scheduler里面提交了一个runnable。最终这个Runnable肯定会执行。也就是上文parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));传入来的SubscribeTask
1 2 3 4 5 6 7 8 9 10 11 12 final class SubscribeTask implements Runnable private final SubscribeOnObserver<T> parent; SubscribeTask(SubscribeOnObserver<T> parent) { this .parent = parent; } @Override public void run () source.subscribe(parent); } }
parent可以直接看到是SubscribeOnObserver对象source是使用外部类的对象。
1 2 3 4 5 6 7 8 9 public ObservableSubscribeOn (ObservableSource<T> source, Scheduler scheduler) super (source); this .scheduler = scheduler; } ... public final Observable<T> subscribeOn (Scheduler scheduler) ObjectHelper.requireNonNull(scheduler, "scheduler is null" ); return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this , scheduler)); }
这里是我们之前分析过的创建ObservableSubscribeOn的代码，这里的this指的是observeOn创建的ObservableObserveOn对象，也就是说，run()里面的代码就是调用了ObservableObserveOn对象的subscribe方法。之前分析过，subscribe()实际上没有做什么，只是调用了subscribeActual()。
1 2 3 4 5 6 7 8 9 10 @Override protected void subscribeActual (Observer<? super T> observer) if (scheduler instanceof TrampolineScheduler) { source.subscribe(observer); } else { Scheduler.Worker w = scheduler.createWorker(); source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize)); } }
observeOn()传递的scheduler不是TrampolineScheduler类型的，所以只需要查看else的代码。这里是先创建了一个工作线程（假设使用的是AndroidScheduler），然后调用了source的subscribe()。需要注意的是这里最后创建了一个ObserveOnOnserver对象。
先看看createWork，在HandlerScheduler。（在AndroidScheduler的DEFAULT中你可以追溯到这里）
1 2 3 4 @Override public Worker createWorker () return new HandlerWorker(handler); }
其中的handler是
1 2 3 4 5 6 7 8 9 10 11 12 13 14 public final class AndroidSchedulers private static final class MainHolder static final Scheduler DEFAULT = new HandlerScheduler(new Handler(Looper.getMainLooper())); } ... public static Scheduler mainThread () return RxAndroidPlugins.onMainThreadScheduler(MAIN_THREAD); } ... }
由上知，返回了一个HandlerWorker对象。再接着看看source.subscribe()，首先这里的source指的是create()方法创建的ObservableCreate对象，调用subscribe()传递的是ObserveOnObserver。看看ObservableCreate的subscribeActual()。
1 2 3 4 5 6 7 8 9 10 11 12 @Override protected void subscribeActual (Observer<? super T> observer) CreateEmitter<T> parent = new CreateEmitter<T>(observer); observer.onSubscribe(parent); try { source.subscribe(parent); } catch (Throwable ex) { Exceptions.throwIfFatal(ex); parent.onError(ex); } }
这里的参数observer是ObserveOnObserver，source是我们代码中创建的oos对象。
首先创建了一个CreateEmitter对象。接着看ObserveOnObserver的onSubscribe方法做了啥：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 @Override public void onSubscribe (Disposable s) if (DisposableHelper.validate(this .s, s)) { this .s = s; if (s instanceof QueueDisposable) { @SuppressWarnings ("unchecked" ) QueueDisposable<T> qd = (QueueDisposable<T>) s; int m = qd.requestFusion(QueueDisposable.ANY | QueueDisposable.BOUNDARY); if (m == QueueDisposable.SYNC) { sourceMode = m; queue = qd; done = true ; actual.onSubscribe(this ); schedule(); return ; } if (m == QueueDisposable.ASYNC) { sourceMode = m; queue = qd; actual.onSubscribe(this ); return ; } } queue = new SpscLinkedArrayQueue<T>(bufferSize); actual.onSubscribe(this ); } }
分析主要代码，就是actual.onSubscribe(this)。actual是构造函数中赋值的，所以我们回到创建ObserveOnObserver的地方，actual指的是SubscribeOnObserver对象。所以它调用了SubscribeOnObserver的onSubscribe()。接下来分析一下它的onSubscribe()
1 2 3 4 5 6 7 8 9 10 SubscribeOnObserver(SingleObserver<? super T> actual, SingleSource<? extends T> source) { this .actual = actual; this .source = source; this .task = new SequentialDisposable(); } @Override public void onSubscribe (Disposable d) DisposableHelper.setOnce(this , d); }
1 2 3 4 5 6 7 8 9 10 11 public static boolean setOnce (AtomicReference<Disposable> field, Disposable d) ObjectHelper.requireNonNull(d, "d is null" ); if (!field.compareAndSet(null , d)) { d.dispose(); if (field.get() != DISPOSED) { reportDisposableSet(); } return false ; } return true ; }
乐观锁
回到主流程上ObservableCreate的subscribeActual()上，source.subscribe(parent);这是最重要的一句代码。source是在构造函数赋值的，看看构造方法：
1 2 3 4 public static <T> Observable<T> create (ObservableOnSubscribe<T> source) { ObjectHelper.requireNonNull(source, "source is null" ); return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source)); }
也就是这里的source就是我们代码中创建的oos对象。parent是新创建的CreateEmitter对象。看看我们oos
1 2 3 4 5 6 7 8 9 10 ObservableOnSubscribe<Integer> oos = new ObservableOnSubscribe<Integer>() { @Override public void subscribe (ObservableEmitter<Integer> e) throws Exception final int max = 100 ; for (int i = 1 ; i <= max; i++) { e.onNext(max); } e.onComplete(); } };
这里就是事件开始的起点，所有事件都由ObservableEmitter开始发送，它是一个接口，在我们的例子中，它的实现类是CreateEmitter，我们分析这个类的onNext()
1 2 3 4 5 6 7 8 9 10 @Override public void onNext (T t) if (t == null ) { onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources." )); return ; } if (!isDisposed()) { observer.onNext(t); } }
1 2 3 4 5 6 7 8 9 10 @Override public void onNext (T t) if (done) { return ; } if (sourceMode != QueueDisposable.ASYNC) { queue.offer(t); } schedule(); }
调用了schedule()
1 2 3 4 5 void schedule () if (getAndIncrement() == 0 ) { worker.schedule(this ); } }
向worker提交了一个Runnable，这里传递的是this。说明它自己肯定实现了这个接口。
1 2 3 4 5 6 7 8 @Override public void run () if (outputFused) { drainFused(); } else { drainNormal(); } }
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 void drainNormal () int missed = 1 ; final SimpleQueue<T> q = queue; final Observer<? super T> a = actual; for (;;) { if (checkTerminated(done, q.isEmpty(), a)) { return ; } for (;;) { boolean d = done; T v; try { v = q.poll(); } catch (Throwable ex) { Exceptions.throwIfFatal(ex); s.dispose(); q.clear(); a.onError(ex); worker.dispose(); return ; } boolean empty = v == null ; if (checkTerminated(d, empty, a)) { return ; } if (empty) { break ; } a.onNext(v); } missed = addAndGet(-missed); if (missed == 0 ) { break ; } } }
a.onNext(v)，a是actual变量，actual又是SubscribeOnObserver对象，查看它的onNext()
1 2 3 4 @Override public void onNext (T t) actual.onNext(t); }
很简单，这里的actual就是我们创建的o了，所以最终调用到了我们的代码里面。
线程切换 subscribeOn()，看ObservableSubscribeOn的代码
1 2 3 4 5 6 7 8 @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))); }
从这里开始就进行了线程的切换，根据上面的分析指导这里是将SubscribeTask作为一个Runnable对象提交进了我们指定的scheduler（subscribeOn传递的）中，所以后面的流程都是在scheduler所在的线程在运行的。
再看observeOn，看ObservableObserveOn的代码
1 2 3 4 5 6 7 8 9 @Override protected void subscribeActual (Observer<? super T> observer) if (scheduler instanceof TrampolineScheduler) { source.subscribe(observer); } else { Scheduler.Worker w = scheduler.createWorker(); source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize)); } }
这里的线程切换是发生在ObserveOnObserver这个对象里面。
1 2 3 4 5 void schedule () if (getAndIncrement() == 0 ){ worker.schedule(this ); } }
schedule，worker.schedule(this)这行代码就发生了线程切换，是将this=作为Runnable对象提交到了我们指定的（observerOn传递的）scheduler中。具体分析，由于之前的流程是在别的线程中，所以想要进行线程切换，最先想到的肯定是Handler。由于我们传递的是AndroidSchedulers.mainThread()，所以我们就分析这个吧。
AndroidSchedulers.mainThread()的实现是HandlerScheduler。看看它的schedule()
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 @Override public Disposable schedule (Runnable run, long delay, TimeUnit unit) if (run == null ) throw new NullPointerException("run == null" ); if (unit == null ) throw new NullPointerException("unit == null" ); if (disposed) { return Disposables.disposed(); } run = RxJavaPlugins.onSchedule(run); ScheduledRunnable scheduled = new ScheduledRunnable(handler, run); Message message = Message.obtain(handler, scheduled); message.obj = this ; handler.sendMessageDelayed(message, Math.max(0L , unit.toMillis(delay))); if (disposed) { handler.removeCallbacks(scheduled); return Disposables.disposed(); } return scheduled; }
而此时的Handler是由new HandlerScheduler(new Handler(Looper.getMainLooper()))传入主线程的Looper，故此时它能切换回主线程中。
总结
每次ObserveOn和subscribeOn的时候，内部都会创建一个新的observable和observer 。
最后我们subscribe的时候，是调用最后创建的observable的方法。而每个observable内部又调用了source的subscribe方法，这样也就形成了一层一层的往前传递的调用链 。当调用到最前面的一个observable的时候，就是我们自己创建的observable，在这里我们需要手动触发该observable对应的observer对象的onNext()。而observer的onNext()内部又调用了actual的onNext()，这样就形成了一层一层往后传递的调用链 。
