RxAndroid
implementation 'io.reactivex.rxjava2:rxandroid:2.0.2'
RxAndroid.png
RxAndroid 是在 RxJava 上的扩展，用法和 RxJava 一样。结构比较简单，主要用在 Android 上的线程切换。
Observable.just(1,2)
.subscribeOn(Schedulers.io())
// 切换到 Android 的主线程
.observeOn(AndroidSchedulers.mainThread())
.subscribe(observerInt)
看源码
public static Scheduler mainThread() {
return RxAndroidPlugins.onMainThreadScheduler(MAIN_THREAD);
}
private static final Scheduler MAIN_THREAD = RxAndroidPlugins.initMainThreadScheduler(
new Callable<Scheduler>() {
@Override public Scheduler call() throws Exception {
return MainHolder.DEFAULT;
}
});
private static final class MainHolder {
static final Scheduler DEFAULT = new HandlerScheduler(new Handler(Looper.getMainLooper()));
}
通过 Looper.getMainLooper() 构建的 Handler 来切换。
看 HandlerScheduler 内部类 HandlerWorker 的 schedule 方法：
@Override
public Disposable schedule(Runnable run, long delay, TimeUnit unit) {
...
run = RxJavaPlugins.onSchedule(run);
ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);
Message message = Message.obtain(handler, scheduled);
message.obj = this; // Used as token for batch disposal of this worker's runnables.
handler.sendMessageDelayed(message, unit.toMillis(delay));
// Re-check disposed state for removing in case we were racing a call to dispose().
if (disposed) {
handler.removeCallbacks(scheduled);
return Disposables.disposed();
}
return scheduled;
}
内部原理就是 Android 的 Handler 这一套机制。
除了切换到主线程还可以根据某个线程上的 Looper 切换到指定线程上。
Looper backgroundLooper = // ...
Observable.just("one", "two", "three", "four", "five")
.observeOn(AndroidSchedulers.from(backgroundLooper))
.subscribe(/* an Observer */)
就是根据传入的 Looper 构建一个 Handler。
public static Scheduler from(Looper looper) {
if (looper == null) throw new NullPointerException("looper == null");
return new HandlerScheduler(new Handler(looper));
}
RxKotlin
implementation 'io.reactivex.rxjava2:rxkotlin:2.2.0'
用 Kotlin 语言写 RxJava 的功能就行，但 Kotlin 本身有些语言特性（比如扩展），使用 RxKotlin 后语法可以更简洁。类似 Google 的 ktx 框架，没什么新功能，只是对一些方法的封装使得使用起来更方便。
RxJava 是调用 Observable 的方法，比如：
Observable.fromArray("a","b","c").subscribe(observerStr)
而使用 RxKotlin 就是这样
arrayOf("a", "b", "c").toObservable().subscribe(observerStr)
源码
fun <T : Any> Array<T>.toObservable(): Observable<T> = Observable.fromArray(*this)
在数组上定义的扩展方法 toObservable() 内部其实就是调用的 fromArray 方法。
其实扩展方法更符合人的思维，不像原来那种，乍看之下是别扭的。
全部的扩展方法见 https://github.com/ReactiveX/RxKotlin。
