前言

通过前面一篇文章，我们应该能大志理解RxJava的原理了吧。

既然RxJava被称为强大的异步处理库，对线程的控制也就不可或缺，今天我们就来谈谈RxJava的线程调度。

RxJava内置线程选项

RxJava中已经内置了很多线程选项供我们选择，其中下面四个已经能满足我们日常的开发。

在RxJava内部使用的是线程池来维护这些线程, 所有效率也比较高。

线程	含义及作用
Schedulers.io()	代表io操作的线程, 通常用于网络,读写文件等io密集型的操作
Schedulers.computation()	代表CPU计算密集型的操作, 例如需要大量计算的操作
Schedulers.newThread()	代表一个常规的新线程
AndroidSchedulers.mainThread()	代表Android的主线程

RxJava线程调度解析

我们第一篇文章所用到的例子都没有用到涉及到对线程的处理，所以上游和下游的工作都是在同一个线程中的。
当我们在主线程中去创建一个上游Observable来发送事件, 则这个上游默认就在主线程发送事件。
当我们在主线程去创建一个下游Observer来接收事件, 则这个下游默认就在主线程中接收事件。

我们来实践一下看看是不是跟我们说的一样：

Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(ObservableEmitter<String> e) throws Exception {
        String s = "1";
        Log.d(TAG, "The observable thread is: " + Thread.currentThread().getName());
        Log.d(TAG, s + " send");
        e.onNext(s);
    }
});
 
Consumer<String> consumer = new Consumer<String>() {
    @Override
    public void accept(String s) throws Exception {
        Log.d(TAG, "The observer thread is: " + Thread.currentThread().getName());
        Log.d(TAG, s + " receive");
    }
};
 
observable.subscribe(consumer);

输出结果：

输出结果

可以看出：
上游Observable指定了3次线程，可是线程保留在第一次指定的newThread中；
下游Observer同样指定了3次线程，每次指定都进行了线程切换，最后以主线程main为指定线程。

实战

说了这么多，栗子呢！！！

我们来模仿一下从通过url从服务器获取图片（子线程耗时操作），再设置到imageView上（主线程更新布局）的例子。

Observable.create(new ObservableOnSubscribe<Drawable>() {
    @Override
    public void subscribe(ObservableEmitter<Drawable> e) throws Exception {
        try {
            Drawable drawable = getDrawable(url);   // 通过url从服务器去获取drawable图片
            if (!e.isDisposed()) {
                if (drawable != null) {
                    e.onNext(drawable);
                }
                e.onComplete();
            }
        } catch (Exception e1) {
            if (!e.isDisposed()) {
                e.onError(e1);
            }
        }
    }
}).subscribeOn(Schedulers.io())                     // 子线程做耗时操作获取图片
    .observeOn(AndroidSchedulers.mainThread())      // 主线程更新UI
    .subscribe(new Consumer<Drawable>() {
        @Override
        public void accept(Drawable drawable) throws Exception {
            mImageView.setImageDrawable(drawable);  // 给imageView设置图片
        }
    }, new Consumer<Throwable>() {
        @Override
        public void accept(Throwable throwable) throws Exception {
            DebugUtil.exception(throwable);
        }
    });

这就是一个Rxjava工作中线程切换的例子啦。

在这里有一个地方要解释一下，我们在调用observable的发送事件前，都加了一个isDisposed()的判断，这其实跟我们上一篇文章提到的Disposable对象有关，我们需要在Activity退出的时候调用dispose()方法，使得下游收不到事件，这样就不会去更新UI了，不然就会出现APP Crash的问题了。