为什么我的RxJava Observable只向第一个消费者发射？

Question

有人能解释一下为什么下面的测试失败了吗？

public class ObservableTest {
    @Test
    public void badObservableUsedTwiceDoesNotEmitToSecondConsumer() {
        // Any simpler observable makes the test pass
        Observable<Integer> badObservable = Observable.just(1)
                .zipWith(Observable.just(2), (one, two) -> Observable.just(3))
                .flatMap(observable -> observable);

        ObservableCalculator calc1 = new ObservableCalculator(badObservable);
        ObservableCalculator calc2 = new ObservableCalculator(badObservable);

        // zipping causes the failure
        // Calling calculate().toBlocking().subscribe() on each calc passes
        // Observable.from(listOfCalcs).flatMap(calc -> calc.calculate()) passes
        Observable.zip(ImmutableList.of(calc1.calculate(), calc2.calculate()), results -> results)
                .toBlocking()
                .subscribe();

        assertThat(calc1.hasCalculated).isTrue();
        assertThat(calc2.hasCalculated).isTrue(); // this fails
    }

    private static class ObservableCalculator {
        private final Observable<?> observable;

        public boolean hasCalculated = false;

        public ObservableCalculator(Observable<?> observable) {
            this.observable = observable;
        }

        public Observable<Void> calculate() {
            return observable.concatMap(o -> {
                hasCalculated = true;
                // returning Observable.just(null) makes the test pass
                return Observable.empty();
            });
        }
    }
}

我试图进一步简化“不好”的观察结果，但找不到任何可以删除的东西来简化它。

然而，我目前的理解是，它是一个可观察的（不管它是如何构造的），应该发出一个值，然后完成。然后，我们基于该可观察对象制作两个类似的对象实例，并在那些使用可观察对象的对象上调用一个方法，记下已经这样做了，然后返回Observable.empty（）。

有人能解释为什么使用这个可观察的会导致测试失败（当使用更简单的可观察的会导致测试通过时）吗？

也可以通过串行调用calculate（）使测试通过。toBlocking（）。subscribe（）而不是使用zip，或使计算返回可见。只是（空）而已。这对我来说有一定的意义（如果calc1为空，zip将不会订阅calc2，因为在这种情况下zip永远不会产生任何结果），但没有完全的意义（我不明白为什么zip在badObservable的更简单版本中不会这样做——calculate（）方法仍然返回空，而不管输入是什么）。

Answer 1

如果你用某个东西压缩一个空的源，操作员会检测到它不能再产生任何值，并取消订阅它的所有源。有一个混合的zip和合并涉及和合并认真对待取消订阅：它不发出值3在所有因此concatMap不调用映射函数的第二个源。