我们都知道android在日常开发时都是在IO线程执行耗时操作，然后在UI线程进行更新UI，那么RxJava怎么进行线程切换的呢？这就用到了subscribeOn和observeOn这两个操作符。首先我们来看下官方文档对着两个操作符的解释：
By default, an Observable and the chain of operators that you apply to it will do its work, and will notify its observers, on the same thread on which its Subscribe method is called.
大致翻译下，默认情况下，一个被观察者和你应用到这条链上的操作符将会执行而且执行完成之后会通知观察者，这些操作都是发生在同一线程中，该线程就是订阅操作被调用的线程，我们来举个例子：
Flowable.just(1)
.map(new Function<Integer, String>() {
@Override
public String apply(Integer integer) throws Exception {
System.out.println(Thread.currentThread().getName());
return String.valueOf(integer);
}
})
.subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
System.out.println(Thread.currentThread().getName());
}
});
可以看到打印的都是在主线程，因为我们订阅的时候是在主线程。
我们在指定下线程加入subscribeOn
Flowable.just(1)
.map(new Function<Integer, String>() {
@Override
public String apply(Integer integer) throws Exception {
System.out.println(Thread.currentThread().getName());
return String.valueOf(integer);
}
})
.subscribeOn(Schedulers.newThread())
.subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
System.out.println(Thread.currentThread().getName());
}
});
image.png
可以看到都是在子线程中执行，这就印证上面的结论。
The SubscribeOn operator changes this behavior by specifying a different Scheduler on which the Observable should operate. The ObserveOn operator specifies a different Scheduler that the Observable will use to send notifications to its observers.
SubscribeOn操作符指定被观察者发送数据的线程，ObserveOn指定被观察者发送通知给观察者的线程，也就是观察者执行操作的线程。
Flowable.just(1)
.map(new Function<Integer, String>() {
@Override
public String apply(Integer integer) throws Exception {
System.out.println(Thread.currentThread().getName());
return String.valueOf(integer);
}
})
.subscribeOn(Schedulers.newThread())
.observeOn(Schedulers.io())
.subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
System.out.println(Thread.currentThread().getName());
}
});
image.png
通过subscribeOn指定发送数据的线程，通过ObserveOn指定观察者接收通知的线程
再来个复杂一点的
Flowable.just(1)
.map(new Function<Integer, String>() {
@Override
public String apply(Integer integer) throws Exception {
System.out.println(Thread.currentThread().getName());
return String.valueOf(integer);
}
})
.subscribeOn(Schedulers.newThread())
.subscribeOn(Schedulers.io())
.subscribeOn(Schedulers.io())
.observeOn(Schedulers.newThread())
.subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
System.out.println(Thread.currentThread().getName());
}
});
image.png
说明调用多次subscribeOn，只有第一次生效。那么多次调用observeOn呢
Flowable.just(1)
.map(new Function<Integer, String>() {
@Override
public String apply(Integer integer) throws Exception {
System.out.println("第一个map执行线程： " + Thread.currentThread().getName());
return String.valueOf(integer);
}
})
.observeOn(Schedulers.io())
.map(new Function<String, String>() {
@Override
public String apply(String s) throws Exception {
System.out.println("第二个map执行线程： " + Thread.currentThread().getName());
return s;
}
})
.subscribeOn(Schedulers.newThread())
.observeOn(Schedulers.io())
.subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
System.out.println("Consumer执行线程： " + Thread.currentThread().getName());
}
});
image.png
observeOn指定了其后代码执行的线程，多次调用则多次切换。第一个map执行的线程是subscribeOn执行的线程，subscribeOn指定的是发射线程，而第一个map前并未通过observeOn在重新指定线程，所以第一个map执行在subscribeOn指定的默认线程RxNewThreadScheduler-1，第二个map上执行了observeOn进行线程切换，所以第二个map执行在RxCachedThreadScheduler-2线程，又因为subscribe之前又调用了observeOn，所以subscribe执行在其上面observeOn指定的线程。
由此可得知，observeOn可达到切换线程的作用，而subscribeOn只是执行了发射数据操作执行的线程。
结尾附上官方文档的一段话和一个图
As shown in this illustration, the SubscribeOn operator designates which thread the Observable will begin operating on, no matter at what point in the chain of operators that operator is called. ObserveOn, on the other hand, affects the thread that the Observable will use below where that operator appears. For this reason, you may call ObserveOn multiple times at various points during the chain of Observable operators in order to change on which threads certain of those operators operate.
网友评论