RxJava.subscribeon（Schedulers.NewThread（））问题

Question

Observable
.from(Arrays.asList("one", "two", "three"))
.doOnNext(word -> System.out.printf("%s uses thread %s%n", word, Thread.currentThread().getName()))
//.subscribeOn(Schedulers.newThread())
.subscribe(word -> System.out.println(word));

它会逐行打印出单词，并与线程的相关信息交织在一起，如预期的那样，线程对于所有下一次调用都是“main”。

但是，当我取消对subscribeON(Schedulers.newThread())调用的注释时，根本不会打印任何内容。为什么不起作用？我希望它为每个onNext()调用启动一个新线程，并通过doonNext()打印该线程的名称。现在，我什么也没看到，其他的调度者也一样。

当我将调用添加到main末尾的thread.sleep(10000L)时，我可以看到输出，这表明RxJava使用的线程都是守护进程。事实是这样的吗？这是否可以改变，但使用自定义ThreadFactory或类似的概念，而不必实现自定义调度器？

通过上述更改，线程名称始终为rxNewThreadScheduler-1，而newThread的文档表示“为每个工作单元创建新的{@link thread}的调度程序”。它不是应该为所有的排放创建一个新线程吗？

Answer 1

正如Vladimir提到的，RxJava标准调度程序在守护进程线程上运行工作，在您的示例中，守护进程线程由于主线程退出而终止。我想强调的是，他们不是在一个新线程上调度每个值，而是在一个新创建的线程上为每个单独的订阅者调度值流。第二次订阅将得到“rxNewThreadScheduler-2”。

您实际上不需要更改默认的调度器，只需用schedulers.from（）包装您自己的基于执行器的调度器，并在需要的地方提供它作为参数：

ThreadPoolExecutor exec = new ThreadPoolExecutor(
        0, 64, 2, TimeUnit.SECONDS, new LinkedBlockingQueue<>());
exec.allowCoreThreadTimeOut(true);

Scheduler s = Schedulers.from(exec);

Observable
.from(Arrays.asList("one", "two", "three"))
.doOnNext(word -> System.out.printf("%s uses thread %s%n", word,
    Thread.currentThread().getName()))
.subscribeOn(s)
.subscribe(word -> System.out.println(word));

我有一系列关于RxJava调度器的博客文章，它们将帮助您实现一个“更永久”的变体。