RxJava平面映射：当一个可观察的结果完成时会发生什么？

Question

我是RxJava的新手，我知道flatmaps用于将发出的项映射到可观察项。我还知道，基于文档，发出的可观测数据都被合并（扁平化）为一个单一的可观测数据流。

我在想，如果这些内在的可观察到的东西都完成了，会发生什么？

例如：我有一个可观察的，它发出一个项数据键。我必须进行另一个异步http调用才能从服务器获取项数据，所以我使用另一个Observable调用它。我使用一个平面地图来连接这两个，并创建一个主要的可观察性。

什么时候调用“SomeMethodThatWantSitems”后面的run（）方法？

public void someMethodThatWantsItems(MyHttpCaller httpCaller, SomeSearchEngine searchEngine)
{
    Consumer<Item> onNextConsumer = 
    Observable<Item> searchObservable = getSearchResult(httpCaller, searchEngine, "The Search Word");
    searchObservable
            .subscribeOn(Schedulers.newThread())
            .subscribe(new Consumer<Item>(){
                           @Override
                           public void accept(@NonNull Item item) throws Exception {
                               //Do stuff with the item
                           }
                       }
                , new Consumer<Exception>() { //some implementation of onErrorConsumer
                    }
                 //OnComplete
                , new Action(){

                        @Override
                        public void run() throws Exception {
                            //When does this get called??? after the search complete or when the first http call is successful? 
                        }
                    });

}

private Observable<String> getSearchResultKeys(SomeSearchEngine searchEngine, String someSearchWord)
{
    return Observable.create(new ObservableOnSubscribe<String>() {
        @Override
        public void subscribe(@NonNull final ObservableEmitter<String> emitter) throws Exception {

            //Assume that our search engine call onFind everytime it finds something
            searchEngine.addSearchListener(new searchEngineResultListener(){
                @Override
                public void onFind(String foundItemKey){
                    emitter.onNext(foundItemKey);
                }

                @Override
                public void onFinishedFindingResults(){
                    emitter.onComplete();
                }
            });

        }
    });
}

private Observable<Item> getItemByKey(MyHttpCaller httpCaller, String key)
{

    return Observable.create(new ObservableOnSubscribe<Item>() {
        @Override
        public void subscribe(@NonNull final ObservableEmitter<Item> emitter) throws Exception {

            //Call the server to get the item
            httpCaller.call(key, new onCompleteListener(){
                @Override
                public void onCompletedCall(Item result)
                {
                    emitter.onNext(result);
                    //The result is complete! end the stream
                    emitter.onComplete();
                }
            });
        }
    });
}

public Observable<Item> getSearchResult(MyHttpCaller httpCaller, SomeSearchEngine searchEngine, String someSearchWord){
    //Where everything comes together
    Observable<String> searchResultObservable = getSearchResultKeys(searchEngine, someSearchWord);
    retuern searchResultObservable
            .observeOn(Schedulers.newThread())
            .flatMap(new Function<String, Observable<Item>>(){
                @Override
                public Observable<Item> apply(String key){
                    return getItemByKey(httpCaller, key);
                }
            });
}

Answer 1

oncomplete()总是调用一次，然后流停止。（这是可观察契约的一部分）。
这意味着在您的情况下，在检索到所有项后，将调用SomeMethodThatWantSitems处的onComplete()。
如果是FlatMap()，完成每个内部可观察性，只需通知源代码可观察性停止从内部可观察性到源代码可观察性的扁平化项，flatmap()只要这个流发送项就会合并内部可观察性的项，因此它基本上将整个内部可观察性流消耗到源流中，整个流直到终止时都是像oncomplete()，所以如果内部可观察性可以发射超过1个项，这意味着