为什么我的RxJava Observable不发出或完成，除非它被阻塞？

Question

我有很多RxJava可观察对象（要么是从Jersey客户端生成的，要么是使用Observable.just（someObject））生成的存根）。它们都应该只发出一个值。我有一个组件测试，模拟了所有Jersey客户端，并使用了Observable。只是（someObject），我看到的行为与运行生产代码时的行为相同。

我有几个类，作用于这些可观察的，执行一些计算（

有一次，在一个这样的类中，我试图压缩我的几个源观测值，然后映射它们——如下所示：

public Observable<Void> doCalculation() {
    return Observable.zip(
        getObservable1(),
        getObservable2(),
        getObservable3(),
        UnifyingObject::new
    ).concatMap(unifyingObject -> unifyingObject.processToNewObservable())
}

// in Unifying Object
public Observable<Void> processToNewObservable() {
    // ... do some calculation ...
    return Observable.empty();
}

然后将所有计算类合并并等待：

// Wait for rule computations to complete
List<Observable<Void>> calculations = ...;
Observable.zip(calculations, results -> results)
        .toBlocking().lastOrDefault(null);

麻烦的是，ProcToNewWatch able（）永远不会执行。通过消除的html" target="_blank">过程，我可以看到这是getWatch able1（）这是个麻烦-如果我用Observable.just（null）替换它，一切都像我想象的那样执行（但是我想要一个真正的空值）。

重申一下，getobserve1（）在生产代码中从Jersey客户端返回一个可观察的，但该客户端是一个Mockito mock，返回可观察的。只是（someValue）在我的测试中。

如果我将getobserve1（）转换为blocking，然后将第一个值包装在just（）中，所有操作都会按照我的想象执行（但我不想引入blocking步骤）：

Observable.zip(
    Observable.just(getObservable1().toBlocking().first()),
    getObservable2(),
    getObservable3(),
    UnifyingObject::new
).concatMap(unifyingObject -> unifyingObject.processToNewObservable())

我的第一个想法是，也许有其他东西正在消耗从我的可观察到的值，并且zip看到它已经完成，从而确定压缩它们的结果应该是一个空的可观察到的。然而，我尝试将. ache（）添加到我认为相关的每个可观察的源上，但这并没有改变行为。

我还尝试在zip之前在getObservable1上添加next/error/complete/finally处理程序（不将其转换为blocking），但它们都没有执行：

getObservable1()
    .doOnNext(...)
    .doOnCompleted(...)
    .doOnError(...)
    .finallyDo(...);

Observable.zip(
    getObservable1(),
    getObservable2(),
    getObservable3(),
    UnifyingObject::new
).concatMap(unifyingObject -> unifyingObject.processToNewObservable())

我对RxJava非常陌生，所以我很确定我错过了一些基本的东西。问题是：我能做什么蠢事？如果从我到目前为止所说的来看，这一点并不明显，那么我能做些什么来帮助诊断这个问题呢？

Answer 1

注意：我在这里没有找到我的答案非常令人满意，所以我深入研究了一下，发现了一个小得多的再现案例，所以我在这里问了一个新问题：为什么我的RxJava可观察只向第一个消费者发出？

我至少已经解决了我的部分问题（向所有试图回答的人道歉，我认为根据我的解释，你没有多大机会）。

执行这些计算的各种类都返回可观察。empty（）（按照我最初的示例中的processToNewObservable（））。据我所知，是可观察的。zip（）不订阅第N个可观察对象，直到第N-1个可观察对象发出一个值，它才会压缩。

我最初的例子声称是getWatch able1（）行为不端——这实际上是轻微的不准确，它是后来在参数列表中可以观察到的。据我所知，让它阻塞，然后将该值转换为可观察值的原因再次起作用是因为让它阻塞和调用首先迫使它执行，我得到了我想要的副作用。

如果我将我所有的计算类更改为返回Observable.just（null），那么一切都会正常工作：所有计算类的可观测对象的最终zip（）都通过它们工作，因此所有预期的副作用都会发生。

从设计的角度来看，返回一个空虚空似乎我肯定做错了什么，但至少这个特殊的问题得到了回答。

Answer 2

根据注释中的响应，其中一个getWatch able不返回任何值，只是完成，或者Mockito mocking出错。下面的独立示例适用于我。你能检查一下它，然后慢慢地改变它，看看哪里坏了吗？

Observable.zip(
        Observable.just(1),
        Observable.just(2),
        Observable.just(3),
        (a, b, c) -> new Integer[] { a, b, c })
 .concatMap(a -> Observable.from(a))
 .subscribe(System.out::println)
 ;

Answer 3

可观察到的物体必须发射才能启动链条。你必须把你的管道想象成一个宣言，当可观察到的东西发射时会发生什么。

你没有分享实际被观察到的东西，而是可以观察到的。just（）使可观察对象立即发射包裹的对象。