rxJava3 中的即时调度器在哪里？答案

【问题标题】：Where is the Immediate Scheduler in rxJava3?rxJava3 中的即时调度器在哪里？
【发布时间】：2021-02-07 09:39:48
【问题描述】：

在rxJava 1 中有Scheduler.immediate()，可让您安排在当前线程上的工作。在rxJava 3我找不到这个调度器了。

有谁知道 rxJava 3 中 Scheduler.immediate() 的替代品是什么？

我的用例：

我有一个客户端 API，用于从远程服务器订阅无限的事件流（例如新闻提要）。 API 通过我注册的回调通知我事件：

Observable.create(emitter -> apiClient.registerCallback(event -> emitter.onNext(event)))
        .observeOn(Schedulers.immediate())   // I'd like downstream operators to run on current thread
        .map(myFunc);

但是，API 从不同的线程调用我的回调。我希望在当前线程（创建Observable的线程）上运行myFunc之类的下游计算，以免阻塞API的线程。

【问题讨论】：

It has been removed in 2.x 因为它的使用非常有限且容易出错。此外，在您的示例中，Observable 没有实际效果，因为myFunc 将始终在执行emitter.onNext 的调用者线程上运行。此外，如果它已经在 Scheduler 或阻塞上，您可能无法返回到“当前线程”。
@akarnokd 谢谢。但是，我不明白为什么这是真的：“myFunc 将始终在执行emitter.onNext 的调用者线程上运行”。对.observeOn(Schedulers.immediate()) 的调用不会将下游操作更改为不同的线程吗？
没有。 Scheduler.immediate() 不会向流中引入任何线程，因此带有它的 observeOn 是无操作的。
@akarnokd 我明白了——然后我想我的问题是：我将如何在创建 observable 的线程上运行.map(myFunc)（通过Observable.create()）？
你的create首先运行的是什么线程？

标签： java rx-java scheduler

【解决方案1】：

AFAIK，在 RxJava 3 中，您可以使用 ImmediateThinScheduler 来获得相同的效果。
虽然它保存在internal 包中，但您可以使用它。

API 非常简单，如果您不想依赖他们的 internal 包，您实际上可以自己创建一个。

【讨论】：