在 RxJava 中，我无法从 flatMap 发出 onComplete答案

【问题标题】：In RxJava, I can't emit a onComplete from a flatMap在 RxJava 中，我无法从 flatMap 发出 onComplete
【发布时间】：2016-01-15 05:26:05
【问题描述】：

当我使用平面地图时，我的 Observable 链没有终止。

我把我的例子归结为：

int count = Observable.just(1,2,3)
        .flatMap(s -> Observable.<Integer>create(subscr-> {
            subscr.onNext(s);
            if(s>2) {
                subscr.onCompleted();
            }
        }))
        .doOnEach(a->System.err.println(a.getKind()+" -> "+a.toString()))
        .count()
        .toBlocking()
        .first();
    System.err.println("Count: "+count);

我希望“doOnEach”报告三个 onNext 事件，然后是一个 onCompleted，最后链应该终止。

但是，输出是这样的：

OnNext -> [rx.Notification@c9e6d24d OnNext 1]
OnNext -> [rx.Notification@c9e6d24e OnNext 2]
OnNext -> [rx.Notification@c9e6d24f OnNext 3]

（然后in一直挂着）

如果我删除 flatMap 运算符：

int count = Observable.just(1,2,3)
        .doOnEach(a->System.err.println(a.getKind()+" -> "+a.toString()))
        .count()
        .toBlocking()
        .first();
System.err.println("Count: "+count);

...它完全按预期工作：

CREATED!
OnNext -> [rx.Notification@e3598bd9 OnNext 1]
OnNext -> [rx.Notification@e3598bda OnNext 2]
OnNext -> [rx.Notification@e3598bdb OnNext 3]
OnCompleted -> [rx.Notification@3834d63f OnCompleted]
Count: 3

我想我做错了什么（我无法想象这种基本场景中的错误），但我没有看到它。

感谢任何帮助...（我使用的是 RxJava 1.1.0）

【问题讨论】：

标签： rx-java

【解决方案1】：

问题在于以下几行：

        if(s>2) {
            subscr.onCompleted();
        }

Observables 和 s == 1 和 s == 2 不会发出 onCompleted。由于flatMap 仅在所有Observables 完成后才发出onCompleted，所以在您的情况下它不会发出onCompleted。

【讨论】：

啊啊啊，我必须完成每一个平面图调用，然后它就会完成。这就解释了。
我也陷入了这个问题。 FlatMap 文档解释了它发生的原因：“FlatMap 运算符通过将您指定的函数应用于源 Observable 发出的每个项目来转换 Observable，其中该函数返回一个本身发出项目的 Observable。然后，FlatMap 合并这些结果的发射Observables，将这些合并的结果作为自己的序列发出。”
“由于 flatMap 仅在所有 Observable 完成时才发出 onCompleted”，+1 小费，救了我的命！