如何解决RxJava中值发射代码和值接收代码的线程执行
我有以下代码:
private static void log(Object msg) {
System.out.println(
Thread.currentThread().getName() +
": " + msg);
}
Observable<Integer> naturalNumbers = Observable.create(emitter -> {
log("Invoked"); // on main thread
Runnable r = () -> {
log("Invoked on another thread");
int i = 0;
while(!emitter.isDisposed()) {
log("Emitting "+ i);
emitter.onNext(i);
i += 1;
}
};
new Thread(r).start();
});
Disposable disposable = naturalNumbers.subscribe(i -> log("Received "+i));
所以这里我们有两个重要的 lambda 表达式。第一个是我们传递给 Observable.create 的,第二个是我们传递给 Observable.subscribe() 的回调函数。在第一个 lambda 中,我们创建一个新线程,然后在该线程上发出值。在第二个 lambda 中,我们有代码来接收第一个 lambda 代码中发出的那些值。我观察到两个代码都在同一个线程上执行。
Thread-0: Invoked on another thread
Thread-0: Emitting 0
Thread-0: Received 0
Thread-0: Emitting 1
Thread-0: Received 1
Thread-0: Emitting 2
Thread-0: Received 2
为什么会这样?默认情况下,RxJava 是否在同一线程上运行代码发出值(可观察)和代码接收值(观察者)?
解决方法
让我们看看,如果您使用 Thread
来执行可运行文件,会发生什么:
测试
@Test
void threadTest() throws Exception {
log("main");
CountDownLatch countDownLatch = new CountDownLatch(1);
new Thread(
() -> {
log("thread");
countDownLatch.countDown();
})
.start();
countDownLatch.await();
}
输出
main: main
Thread-0: thread
看来,主入口点是从 main
线程调用的,而新创建的 Thread
被称为 Thread-0
。
为什么会这样?默认情况下,RxJava 是否在同一线程上运行代码发出值(可观察)和代码接收值(观察者)?
默认情况下 RxJava
是单线程的。因此,如果生产者没有通过 observeOn
、subscribeOn
或不同的线程布局进行不同的定义,则将在 consumer
(订阅者)线程上发出值。这是因为 RxJava
默认运行订阅堆栈上的所有内容。
示例 2
@Test
void fdskfkjsj() throws Exception {
log("main");
Observable<Integer> naturalNumbers =
Observable.create(
emitter -> {
log("Invoked"); // on main thread
Runnable r =
() -> {
log("Invoked on another thread");
int i = 0;
while (!emitter.isDisposed()) {
log("Emitting " + i);
emitter.onNext(i);
i += 1;
}
};
new Thread(r).start();
});
Disposable disposable = naturalNumbers.subscribe(i -> log("Received " + i));
Thread.sleep(100);
}
输出 2
main: main
main: Invoked
Thread-0: Invoked on another thread
Thread-0: Emitting 0
Thread-0: Received 0
Thread-0: Emitting 1
在您的示例中,很明显,主方法是从主线程调用的。此外,subscribeActual
调用也在调用线程 (main
) 上运行。但是 Observable#create
lambda 从新创建的线程 onNext
调用 Thread-0
。该值从调用线程推送给订阅者。在这种情况下,调用线程是 Thread-0
,因为它在下游订阅者上调用 onNext
。
如何区分生产者和消费者?
使用 observeOn
/ subscribeOn
运算符来处理 RxJava
中的并发。
我应该使用低级线程构造ẁith RxJava吗?
不,您不应该使用 new Thread
来将生产者与消费者分开。很容易破坏契约,onNext
不能被并发调用(交错),因此破坏契约。这就是 RxJava
提供带有 Scheduler
的名为 Worker
的结构以减少此类错误的原因。
注意:
我认为这篇文章很好地描述了它:http://introtorx.com/Content/v1.0.10621.0/15_SchedulingAndThreading.html。请注意这是 Rx.NET,但原理完全相同。如果您想阅读有关 RxJava
的并发性,您还可以查看 Davids 博客 (https://akarnokd.blogspot.com/2015/05/schedulers-part-1.html) 或阅读本书(Reactive Programming with RxJava https://www.oreilly.com/library/view/reactive-programming-with/9781491931646/)
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。