在不指定线程的情况下, RxJava 遵循的是线程不变的原则,即:在哪个线程调用 subscribe(),就在哪个线程生产事件;在哪个线程生产事件,就在哪个线程消费事件。如果需要切换线程,就需要用到 Scheduler (调度器)。
可以使用subscribeOn()指定观察者代码运行的线程,使用observerOn()指定订阅者运行的线程。
RxJava内置的几种线程调度器
有了以上这几个 Scheduler ,就可以使用 subscribeOn() 和 observeOn() 两个方法来对线程进行控制了。
注意:observeOn() 指定的是 Subscriber 的线程,而这个 Subscriber 并不一定是 subscribe() 参数中的 Subscriber(这块参考RxJava变换部分),而是 observeOn() 执行时的当前 Observable 所对应的 Subscriber ,即它的直接下级 Subscriber 。换句话说,observeOn() 指定的是它之后的操作所在的线程。因此如果有多次切换线程的需求,只要在每个想要切换线程的位置调用一次 observeOn() 即可。
看个例子:
Observable.just(1, 2, 3, 4)
.subscribeOn(Schedulers.io()) // 指定 subscribe() 发生在 IO 线程
.observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回调发生在主线程
.subscribe(new Action1<Integer>() {
@Override
public void call(Integer number) {
Log.e("normal", "number:" + number);
}
});
最普通用法,长出现在Retrofit+RxJava进行网络请求,在io线程中请求,在主线程中接收处理。
下面的实例,在Observable.OnSubscribe的call()中模拟了长时间获取数据过程,在Subscriber的noNext()中显示数据到UI。
Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("info1");
SystemClock.sleep(2000);
subscriber.onNext("info2-sleep 2s");
SystemClock.sleep(3000);
subscriber.onNext("info2-sleep 3s");
SystemClock.sleep(5000);
subscriber.onCompleted();
}
})
.subscribeOn(Schedulers.io()) //指定 subscribe() 发生在 IO 线程
.observeOn(AndroidSchedulers.mainThread()) //指定 Subscriber 的回调发生在主线程
.subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {
Log.v(TAG, "onCompleted()");
}
@Override
public void onError(Throwable e) {
Log.v(TAG, "onError() e=" + e);
}
@Override
public void onNext(String s) {
showInfo(s); //UI view显示数据
}
});
值得注意:subscribeOn () 与 observeOn()都会返回了一个新的Observable,因此若不是采用上面这种直接流方式,而是分步调用方式,需要将新返回的Observable赋给原来的Observable,否则线程调度将不会起作用。
不同于 observeOn() , subscribeOn() 的位置放在哪里都可以,但它是只能调用一次的。
对于Scheduler调度器的简单使用,大致就介绍到这里,更多精彩内容欢迎关注我的微信公众号——Android机动车