ReactiveX 系列文章目录

---

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。