RxJava 中的 'Subscriber Thread' 是什么意思

Question

我在读一本关于 RxJava 的书，下面是其中的摘录：

Observable.create(s -> {
   ... async subscription and data emission ...
})
.doOnNext(i -> System.out.println(Thread.currentThread()))
.filter(i -> i % 2 == 0)
.map(i -> "Value " + i + " processed on " + Thread.currentThread())
.subscribe(s -> System.out.println("SOME VALUE =>" + s));
System.out.println("Will print BEFORE values are emitted")”

他写道，由于操作是异步的，值将在不同于订阅者线程的线程上发出。由于订阅者的线程是非阻塞的，因此 Will print BEFORE values are emitted 将在任何值被推送到订阅者之前打印出来。

我无法理解这个订阅者线程是什么？我能以某种方式得到它的名字吗？为了测试作者在说什么，我写了一些代码：

Observable.create(
            subscriber -> {
              System.out.println("Entering thread: " + Thread.currentThread().getName());
              Thread t =
                  new Thread(
                      new Runnable() {
                        @Override
                        public void run() {
                          try {
                            System.out.println(
                                "Emitting threads: " + Thread.currentThread().getName());
                            Thread.sleep(5000);
                            subscriber.onNext(1);
                            subscriber.onNext(2);
                            subscriber.onNext(3);
                            subscriber.onCompleted();
                          } catch (InterruptedException e) {
                            subscriber.onError(e);
                          }
                        }
                      });
              t.start();
            })
        .subscribe(
            a -> {
              System.out.print("Subscriber thread: " + Thread.currentThread().getName() + " ");
              System.out.println(a);
            });
    System.out.println("Main thread exiting:");

当我运行上述代码时，我发现订阅者线程与调用 onNext() 的线程相同。由于这是一个异步操作，因此项目应该在与订阅者线程不同的线程上发出。怎么两个线程都一样？

此外，即使在主线程退出后，程序仍保持运行ning 并仅在通过 onNext() 将所有元素下推后才终止。为什么会这样？为什么程序没有退出？

Answer 1

默认情况下，Rx 是单线程的，这意味着 Observable 和我们可以应用于它的运算符链将在其 subscribe() 方法所在的同一线程上通知其观察者被调用。

Answer 2

订阅者线程是调用 subscribe 的线程。由于 RxJava 流是这样的订阅调用链，这些调用可以通过 subscribeOn.

移动到序列中的其他线程。

例如：

Observable.fromCallable(() -> Thread.currentThread())
.subscribe(System.out::println);

System.out.println("Subscriber thread: " + Thread.currentThread());

这将打印订阅者线程两次。或者

Observable.fromCallable(() -> Thread.currentThread())
.subscribeOn(Schedulers.single())
.subscribe(System.out::println);

System.out.println("Subscriber thread: " + Thread.currentThread());

Thread.sleep(1000);

这将打印订阅者线程和 RxSingleScheduler 线程。

RxJava 中的 'Subscriber Thread' 是什么意思

What Is Meant By 'Subscriber Thread' in RxJava

java

reactive-programming

rx-java