RxJava Flowable 缓存到单一死锁

Question

下面是我的代码片段。我知道你不应该cachedFlowable这样阻止，但这只是一个例子。它卡在了线上blockingGet。如果我替换singleOrError为singleElement，代码仍然会卡住。如果我替换singleOrError为firstElement，代码将不再卡住。有人可以向我解释为什么会出现这种情况吗？    public static void main(String[] args) {        final Flowable<Integer> cachedFlowable = Flowable.just(1).cache();        cachedFlowable                .doOnNext(i -> {                    System.out.println("doOnNext " + i);                    final Integer j = cachedFlowable.singleOrError().blockingGet();                    System.out.println("after blockingGet " + j);                })                .blockingSubscribe();    }

Answer 1

它与运算符死锁的原因singleX是，此类运算符等待可能的第二项发射，但由于您阻止了它们，因此来自主源的任何第二项或完成都无法执行。由于firstX他们只关心第一个项目，因此几乎立即解锁，从而允许源完成。

所以是的，您不应该在这样的流程中使用阻塞方法，而应该使用flatMaporconcatMap执行每个项目的子流程：

var cache = Flowable.just(1).cache();

cache

.doOnNext(i -> System.out.println("doOnNext " + i))

.concatMapSingle(item -> cache.firstOrError())

.doOnNext(j -> System.out.println("after " + j))

.blockingSubscribe();