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/）

java java rx-java3