RxJava observeOn调度程序线程查询

Question

我必须根据传入的请求写入文件。由于多个请求可能同时出现，我不希望多个线程试图一起覆盖文件内容，这可能会导致丢失一些数据。

因此，我尝试使用实例变量PublishSubject收集所有请求的数据。我在初始化期间订阅了publishSubject，此订阅将在应用程序的整个生命周期内保持不变。此外，我还在一个单独的线程（由Vertx事件循环提供）上观察到相同的实例，该线程调用负责编写文件的方法。

private PublishSubject<FileData> publishSubject = PublishSubject.create();

private void init() {
    publishSubject.observeOn(RxHelper.blockingScheduler(vertx)).subscribe(fileData -> writeData(fileData));
}

稍后在请求处理过程中，我调用onNext，如下所示：

handleRequest() {
   //do some task
   publishSubject.onNext(fileData);
}

我知道，当我调用onNext时，数据将排队，由操作员指定的特定线程写入文件。然而，我想了解的是

1. 此线程是否仅针对此任务在等待状态下被阻止？或者，
2. 当没有文件写入时，它还会用于其他活动吗？我不想让vertx事件循环中的一个线程在等待状态中浪费掉，从而使用这种方法。此外，如果有更好的方法，请提出建议

提前谢谢。

Answer 1

实际上，RxJava会为您做到这一点，根据定义，排放将以串行方式进行：

观察者必须连续（而不是并行）向观察者发出通知。它们可能会从不同的线程发出这些通知，但在通知之间的关系之前必须有一个正式的事件发生。（可观察合同）

因此，只要您在订阅服务器上的onNext（）中运行阻塞调用（并且不会手动将工作转移到其他线程），您就可以了，不会发生并行写入。

实际上，你的担忧应该来自相反的方向——背压
您应该在这里选择背压策略，好像请求会更快，然后您会处理它们（写入文件），您可能会溢出缓冲区并陷入麻烦。（考虑使用Flowable，并根据需要选择背压策略。

关于您的问题，这取决于调度器，您使用的是RxHelper。blockingScheduler（vertx）这似乎是您的自定义代码，所以我无法判断，如果调度程序以工作队列方式使用共享线程，那么它将不会保持空闲<无论如何，Rx不会为您确定这一点，调度器的责任是根据其逻辑将工作分配给某个线程。