RxJava2 subscribe会在一段时间后停止观察，但在flowable完成后会继续

Question

我很难理解以下代码示例的行为；

    Flowable<String> f = Flowable.just(1)
            .flatMap(it -> Flowable.create(e -> {

                for(int i = 1; i < 1001; ++i) {
                    log.info("Emitting: " + i);
                    if(i % 10 == 0) {
                        Thread.sleep(1000);
                    }
                    e.onNext(i);
                }

                e.onComplete();
            }, BackpressureStrategy.BUFFER))
            .map(String::valueOf)
            .subscribeOn(Schedulers.io())
            .observeOn(Schedulers.newThread());

    f.subscribe(val -> {
        Thread.sleep(100);
        log.info("Observing: " + val);
    });

    Thread.sleep(1000000);

代码正常工作，直到subscribe调用观察到128项。发射和观察是平行的。但在那之后，Flowable继续发射项目（显然在某处排队），但直到所有1000个项目发射完毕，才观察到任何项目。发射完所有1000个项目后，剩下的项目(

这看起来与128的背压缓冲区大小有关，但我仍然希望发射和观察在整个1000个项目中是平行的，因为观察者显然不比发射者慢。我有什么遗漏吗？我应该怎么做来修复代码？

Answer 1

这是由于create和subscribeOn之间存在相同的池死锁造成的：

如果链中有create（FlowableOnSubscribe，BackPressureStragy）类型source up，建议使用subscribeOn（scheduler，false）来避免相同的池死锁，因为请求可能会堆积在急切/阻塞的发射器后面。

//...
.subscribeOn(Schedulers.io(), false)
//...

编辑：

我通过替换Flowable尝试了最初的示例（以及您建议的修复）。创建一个可流动的。但我没有遇到任何问题。你能举一个可能出现问题的例子吗？

Flowable.range(1, 10)
    .subscribeOn(Schedulers.io(), false)
    .doOnNext(v -> System.out.println(Thread.currentThread().getName()))
    .observeOn(Schedulers.single(), false, 1)
    .blockingSubscribe();

这将首先打印RxCachedThreadScheduler-1，然后打印RxSingleScheduler-19次，因为观察的补给请求将在单个调度程序上运行，而不是路由回io调度程序。请尝试使用SubbeOn true。