为何在嵌套流中使用computation()调度器会引发死锁？

最近在编写基于RxJava的复杂流逻辑时，踩了个巨头疼的死锁坑——特定场景下必现，查了快一下午才揪出问题，还整理了个极简的复现示例，给碰到类似问题的朋友参考：

public static void main(String[] args) throws InterruptedException {
    Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9)
        .flatMap(x -> Observable.fromIterable(produceMultiple(x)))
        .subscribeOn(Schedulers.computation())
        .subscribe(System.out::println);
    Thread.sleep(50_000);
}

private static List<Integer> produceMultiple(int x) {
    // 模拟阻塞/耗时操作，这是触发死锁的关键诱因
    try {
        Thread.sleep(100);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
    return List.of(x, x * 2);
}

问题根源分析 #

为什么这个看似简单的流会触发死锁？核心在于**subscribeOn的位置和computation线程池的特性**：

• subscribeOn(Schedulers.computation())指定了整个流的订阅逻辑（包括flatMap内部的事件处理）都运行在computation线程池里，而这个线程池的核心线程数默认等于CPU核心数，是固定大小的。
• 当flatMap同时触发多个produceMultiple调用时，每个调用都会阻塞当前的computation线程。很快所有computation线程都会被这种阻塞操作占满，RxJava再也没有剩余线程来处理后续的流事件、完成订阅逻辑，整个流彻底卡死——这就是典型的线程耗尽型死锁。

可行的解决方案 #

针对这类问题，我整理了几个有效的解决思路：

• 调整subscribeOn的位置：如果上游的事件发射不需要computation线程，把subscribeOn移到flatMap之前，让flatMap内部的逻辑能更合理地利用线程：

  Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9)
      .subscribeOn(Schedulers.computation())
      .flatMap(x -> Observable.fromIterable(produceMultiple(x)))
      .subscribe(System.out::println);
  

• 换用适合阻塞操作的线程池：computation线程池是专为CPU密集型任务设计的，不适合处理阻塞操作。如果produceMultiple里有IO或阻塞逻辑，改用Schedulers.io()（动态扩容的线程池，专为IO密集型场景优化）：

  Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9)
      .flatMap(x -> Observable.fromIterable(produceMultiple(x))
          .subscribeOn(Schedulers.io())) // 给flatMap内部的阻塞逻辑单独指定IO线程
      .subscribeOn(Schedulers.computation())
      .subscribe(System.out::println);
  

• 把阻塞操作异步化：尽量避免在流的处理线程里直接执行阻塞逻辑，用Observable.fromCallable包裹阻塞任务并指定合适的线程池，让阻塞逻辑不占用流的核心处理线程：

  Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9)
      .flatMap(x -> Observable.fromCallable(() -> produceMultiple(x))
          .subscribeOn(Schedulers.io())
          .flatMapIterable(list -> list))
      .subscribeOn(Schedulers.computation())
      .subscribe(System.out::println);
  

内容的提问来源于stack exchange，提问作者Soamid