接收从 Observable 发出的项目答案

【问题标题】：Receiving items emitted from an Observable接收从 Observable 发出的项目
【发布时间】：2018-11-21 12:50:26
【问题描述】：

这是我的代码：

  public static void main(String[] args) {
    Observable.just("747", "737", "777")
        .flatMap(
            a -> {
              try {
                Thread.sleep(5000);
              } catch (InterruptedException e) {
                e.printStackTrace();
              }
              return Observable.just(a).subscribeOn(Schedulers.newThread());
            })
        .subscribe(p -> System.out.println("Received " + p + " on thread " + Thread.currentThread().getName()));
  }

据我了解，observable 的每个项目都应该在单独的线程中运行（确实会发生），并且结果将被发送到执行工作的同一线程（也会发生）。但我无法理解的是，为什么main 线程没有退出并等待后台线程完成？只要每个后台线程都在运行，程序就会继续运行。

【问题讨论】：

尝试检查是否有线程在后台运行，并增加Thread.sleep(5000);以及减少后台线程中的sleep。
在.subscribe(...) 之后添加.dispose() 会发生什么？ :)
@i_th 唯一运行的后台线程是主线程。

标签： java multithreading rx-java reactive-programming

【解决方案1】：

如果您查看线程转储，您会发现main 线程实际上卡在sleep 语句上。这就是它没有退出的原因。

这是因为它是执行flatMap 运算符的线程，所以它被卡住了。这也是代码需要很长时间才能运行的原因。您可以通过在sleep 之前插入打印语句来轻松验证它：

try {
  System.out.println(Thread.currentThread().getName() + " is sleeping");
  Thread.sleep(5000);
} catch (InterruptedException e) {
  e.printStackTrace();
}

输出是这样的：

main is sleeping
main is sleeping
Received 747 on thread RxNewThreadScheduler-1
main is sleeping
Received 737 on thread RxNewThreadScheduler-2
Received 777 on thread RxNewThreadScheduler-3

我想你的本意是这样写的：

System.out.println(Thread.currentThread().getName()  + " is creating the observable");
Observable.just("747", "737", "777")
        .flatMap(a ->
                 Observable.fromCallable(() -> {
                  try {
                    System.out.println(Thread.currentThread().getName() + " is sleeping");
                    Thread.sleep(5000);
                  } catch (InterruptedException e) {
                    e.printStackTrace();
                  }
                  return a;
                }).subscribeOn(Schedulers.newThread())
      ).subscribe(p -> System.out.println("Received " + p + " on thread " + Thread.currentThread().getName()));
System.out.println(Thread.currentThread().getName() + " is going to exit");

输出：

main is creating the observable
main is going to exit
RxNewThreadScheduler-3 is sleeping
RxNewThreadScheduler-2 is sleeping
RxNewThreadScheduler-1 is sleeping
Received 777 on thread RxNewThreadScheduler-3
Received 747 on thread RxNewThreadScheduler-1
Received 737 on thread RxNewThreadScheduler-1

在此版本中，main 在 Observable 创建后立即退出。

【讨论】：