我很难理解下面代码示例的行为:
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)将一次性被观察到。这似乎与后压缓冲区大小为128有关,但我仍然期望发射和观察在整个1000个项目中都是并行的,因为观察者显然不比发射器慢。这里有什么我漏掉的东西吗?我该怎么修改代码呢?
Flowable.create
,并且会干扰其他源,因为对于它们来说,subscribeOn 必须将发射固定到特定的调度程序。使用命令式的create
,事实上你正在“阻塞”发射线程,因此没有必要也没有能力回到由下游请求触发的那个线程。 - akarnokdsubscribeOn(Schedulers.io(), false)
可能会与其他流产生问题,对吗? 我有一个共同的地方,在那里我有这个subscribeOn
调用,并且我的来源可能不同,有些是Flowable.create
,有些则不是。我尝试了原始示例(加上您建议的修复),将Flowable.create
替换为Flowable.range
,但我没有遇到任何问题。您能否举个例子说明什么情况下可能会出现问题? - Ertan D.doOnNext
之前使用observeOn
时,无论subscribeOn
是true还是false,它都会以相同的方式工作。在我的使用案例中,我总是先调用subscribeOn
,然后立即调用observeOn
。因此,对于我的使用情况,你的修复看起来不错,希望不会对没有Flowable.create
的源代码造成任何问题。 - Ertan D.