为什么在嵌套流中使用computation（）调度器会导致死锁？

Question

最近我在写一些复杂的基于RX的流程，发现它总是在特定情况下产生死锁。我花了几个小时才找出问题所在，似乎可以在这个简单的示例中重现：

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) {
    return Observable.range(1, x)
        .flatMap(y -> Observable.fromCallable(() -> {
                Thread.sleep(1000);
                return 10 * x + y;
            }).subscribeOn(Schedulers.computation())
        ).toList().blockingGet();
}

此程序应打印以下值：11、21、22、31、32、33、。。。，通常，值可以表示为XY。每组X中的值的顺序可以是随机的，但组的顺序应该是升序。如果previous仍在计算，则不应发出新组（这是我的原始情况）。

问题是，如果您运行这段代码，您将只看到前几个元素的输出-我认为它与处理器的数量相关，因为computation（）调度器使用固定的线程池大小。

你知道为什么会这样吗？这看起来很奇怪，因为主链的数字（1，2，3，…）在池中的单个线程上处理，并且每次produceMultiple使用blockingGet（）完成其作业时，其他线程都应该是空闲的。

如果您修改任何提供不同调度器的subscribeOn（）（或者一个是computation（），另一个不同），一切正常。此外，如果我从自定义线程池执行器（绑定到4个线程）创建调度程序，它仍然可以工作！

Answer 1

不要在计算调度器上使用blockingGet，因为您可能会阻止代码其他部分使用的支持线程。您甚至不必在代码中使用它，只需从produceMultiple返回可观察的：

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

    Thread.sleep(50_000);
}

static Observable<Integer> produceMultiple(int x) {
    return Observable.range(1, x)
        .flatMap(y -> Observable.fromCallable(() -> {
                Thread.sleep(1000);
                return 10 * x + y;
            }).subscribeOn(Schedulers.computation())
        );
}