rx-java - 在 RxJava 映射中阻塞无法完成

Question

以下(使用 RxJava 1.2.4 的错误方式)代码无法解锁并且永远不会完成。

Scheduler scheduler = Schedulers.computation();
Observable.range(0, 100).map(i -> {
    System.out.println("onNext " + i);
    return Observable.just(i).subscribeOn(scheduler).toBlocking().single();
}).subscribeOn(scheduler).toBlocking().subscribe();
System.out.println("finished");

如果将第一行更改为固定线程池，它就会结束。

Scheduler scheduler = Schedulers.from(Executors.newFixedThreadPool(8));

导致第一个示例无法运行的计算调度程序有何特殊之处？

Answer 1

不要那样做

请注意有关计算调度程序的文档说明:

This can be used for event-loops, processing callbacks and other computational work. Do not perform IO-bound work on this scheduler.

他们想说:不要在这个调度器上执行任何阻塞事件。

所以你这样做是违法的，但它是一个很好的示范。

为什么会发生死锁

同样的死锁发生在RxJava 2(此时是2.0.4)

它的发生是由于 computation 调度器的实现方式。它创建固定数量的单线程 worker (这些 worker 的数量是 CPU 内核的数量；在我的例子中是 4 个)。它将任务分配给这些工作人员的方式很简单轮询。现在让我们看看在您的示例中将哪些任务分配给了哪些工作人员。

• worker 1 <- subscribe() 调用调用一个循环来生成 range 中的整数；请注意，直到所有值都传递到下游，此任务才会完成
• worker 2 <- just(0)...toBlocking().single() 用于第一个生成的整数；由于该值已经可用，因此无需真正阻塞即可立即完成
• worker 3 <- just(1)...toBlocking().single() 用于第二个生成的整数；这个立即完成
• worker 4 <- just(2)...toBlocking().single() 用于第三个生成的整数；这个立即完成

此时我们有 worker 1 仍在忙于执行 range 循环，worker 2-4 空闲。下一个任务来自循环，根据round-robin分配给worker1:

• worker 1 <- just(3)...toBlocking().single() 用于第四个生成的整数；这一个进入队列，而 worker 1 循环卡在等待其结果。这是死锁。

FixedThreadPool 调度程序不会锁定，因为它会将任务分配给可用线程，而不是以循环方式。只需确保它有 1 个以上的线程。

阻塞是邪恶的

通常，您应该避免在 Rx 管道中进行阻塞操作。 Rx 提供了很好的工具来执行异步任务而不阻塞。您可以使用 flatMap 代替 map，例如:

Scheduler scheduler = Schedulers.computation();
Observable.range(0, 100).flatMap(i -> {
    System.out.println("onNext " + i);
    return Observable.just(i).subscribeOn(scheduler);
}).subscribeOn(scheduler).toBlocking().subscribe();
System.out.println("finished");

即使使用单线程调度程序，这也可以工作。

除了 Observable.just(i)，您还可以像这样调用真正的异步任务 Observable.fromFuture(asyncService(i))。

如果有必要，请使用 concatMap 而不是 flatMap 来保留项目的顺序。