RxJava 线程模型
本文基于 RxJava 2.1.2 。根据代码和输出日志会更容易理解。
RxJava 的线程模型如下：
1. 不指定线程的情况
-
不指定线程也就是不使用
observeOn和subscribeOn，所有操作在调用subscribe的线程执行。
@Test
public void noThread() {
buildObservable().subscribe();
}
上面代码的输出为：
Thread[main] execute Action start emmit Thread[main] execute Operation-1, event: 1 Thread[main] execute Operation-2, event: 1
2. subscribeOn
-
subscribeOn不管调用多少次，只以第一次为准。如果只使用了subscribeOn、没有使用observeOn，则所有操作在第一次调用生成的线程里执行。
@Test
public void subscribeOn() throws InterruptedException {
CountDownLatch latch = new CountDownLatch(1);
Observable<Integer> observable = buildObservable();
observable
.subscribeOn(scheduler("subscribeOn-1"))
.subscribeOn(scheduler("subscribeOn-2"))
.subscribe(i -> {
showMessageWithThreadName("Action subscribe");
latch.countDown();
});
latch.await();
}
上面代码的输出为：
create scheduler subscribeOn-2 create scheduler subscribeOn-1 Thread[subscribeOn-1] execute Action start emmit Thread[subscribeOn-1] execute Operation-1, event: 1 Thread[subscribeOn-1] execute Operation-2, event: 1 Thread[subscribeOn-1] execute Action subscribe
3. observeOn
-
observeOn必须跟subscribeOn一起使用，单独使用会抛出空引用异常。 -
observeOn应在subscribeOn的后面调用，否则会出现死锁的情况。 -
observeOn操作会更改后续操作的执行线程，直至下一个observeOn调用之前的操作或subscribe操作。
@Test
public void observeOn() throws InterruptedException {
CountDownLatch latch = new CountDownLatch(1);
Observable<Integer> observable = buildObservable();
observable
.subscribeOn(scheduler("subscribeOn-1"))
.observeOn(scheduler("observeOn-1"))
.doOnNext(i -> {
showMessageWithThreadName("Operation-3, event: " + i);
})
.observeOn(scheduler("observeOn-2"))
.subscribe(i -> {
showMessageWithThreadName("subscribe " + i);
latch.countDown();
});
latch.await();
}
上面代码的输出为：
create scheduler subscribeOn-1 Thread[subscribeOn-1] execute Action start emmit Thread[subscribeOn-1] execute Operation-1, event: 1 Thread[subscribeOn-1] execute Operation-2, event: 1 create scheduler observeOn-1 Thread[observeOn-1] execute Operation-3, event: 1 create scheduler observeOn-2 Thread[observeOn-2] execute subscribe 1
4. 辅助代码
// 返回用给定线程名 命名的Scheduler
private Scheduler scheduler(String name) {
return Schedulers.from(Executors.newSingleThreadExecutor(new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
System.out.println("create scheduler " + name);
Thread t = new Thread(r, name);
return t;
}
}));
}
// 输出当前线程名和给的消息
private void showMessageWithThreadName(String msg) {
Thread t = Thread.currentThread();
System.out.printf("%-10s execute %s/n", "Thread[" + t.getName() + "]", msg);
}
// 构建一个带有两个中间操作的 Observable
private Observable<Integer> buildObservable() {
return Observable.fromPublisher((Subscriber<? super Integer> s) -> {
showMessageWithThreadName("Action start emmit");
// 消息源
s.onNext(1);
s.onComplete();
})
.doOnNext(i -> {
showMessageWithThreadName("Operation-1, event: " + i);
})
.doOnNext(i -> {
showMessageWithThreadName("Operation-2, event: " + i);
});
}
正文到此结束
热门推荐
相关文章
近期评论
-
666
-
文章和留言都翻到11页了 没有OOM
-
-
朋友，翻页到11页，及以后，会出现OOM，无法访问
-
-
-
-
版本号是多少，你可以下载哪个代码仓库，jdk选1.8 直接跑就行
-
-
Loading...
![[HBLOG]公众号](https://www.liuhaihua.cn/img/qrcode_gzh.jpg)
