为什么 toMap 的错误放置会导致 RXJava 出现问题？

Question

更新：我正在使用 RxJava 1.x这是以下代码段：private static void tryObservableToMap() {    bad();    good();}private static void good() {    System.out.println("GOOD CASE");    String goodOutput =            m(m(m(m(m(Observable.from(ImmutableList.of("a","b","c","d")), "list")            .distinct(), "distinct")            .flatMap(s ->                    m(m(Observable.fromCallable(() -> getIntForString(s)).subscribeOn(Schedulers.io()), "getInt " + s)                            .map(intValue -> Pair.of(s, intValue)), "pair " + s)), "flatMap")            .toMap(Pair::getKey, Pair::getValue), "toMap")            .map(map -> map.entrySet().stream().map(e -> e.getKey() + ": " + e.getValue()).collect(Collectors.joining("\n"))), "OUTER")            .toBlocking()            .first();    try {        Thread.sleep(2000);    } catch (InterruptedException e) {        e.printStackTrace();    }    System.out.println("\nOutput:");    System.out.println(goodOutput);}private static void bad() {    System.out.println("BAD CASE");    String badOutput =            m(m(m(m(Observable.from(ImmutableList.of("a","b","c","d")), "list")            .distinct(), "distinct")            .flatMap(s ->                    m(m(m(Observable.fromCallable(() -> getIntForString(s)).subscribeOn(Schedulers.io()), "getInt " + s)                            .map(intValue -> Pair.of(s, intValue)), "pair " + s)                            .toMap(Pair::getKey, Pair::getValue), "toMap " + s)), "flatMap")            .map(map -> map.entrySet().stream().map(e -> e.getKey() + ": " + e.getValue()).collect(Collectors.joining("\n"))), "OUTER")            .toBlocking()            .first();    try {        Thread.sleep(2000);    } catch (InterruptedException e) {        e.printStackTrace();    }    System.out.println("\nOutput:");    System.out.println(badOutput);}好与坏的区别在于，对于坏版本，我是.toMap在内部调用.flatMap而不是在.flatMap.如果您运行此代码，您将看到作为执行一部分的所有 observable 的不同事件。我想知道为什么“外部”可观察对象永远不会因坏情况而终止。对RX有更深了解的人可以解释一下吗？

Answer 1

该RXLOG OUTER observable complete丢失是因为有之间的竞争toBlocking().first()和它上面完成的来源。它可能会过早取消订阅，因此上面的源可能没有机会发出onCompleted. 在我的 i7 4770K 上，他们从不completed为我打印。如果您替换first为toIterable().iterator().next()，它将提供必要的机会，您应该始终看到丢失的日志。