Rxjava2 - 如何让观察者在第一个之后停止发射项目或使用 zipWith？

Question

下面的代码可以很好地进行网络调用。但它不断发出相同的结果。我只想获得第一个结果并停止排放。没有命令我可以说只是发出第一个命令。我尝试了 take(1) 但由于某种原因改变了结果大小。

//class variables
  FeedsModel feedsModelResult;
    HashMap<Integer, ProductModel> productMap;

//method

    @Override
       protected Observable buildUseCaseObservable() {
           /* gets feedModel then parses through each feed for product IDs. then does a network call to get each product. stores retrieved
           product model in hashmap for quick retrieval. returns a pair.
            */
           return feedRepository.fetchFeeds(shopId, langId, skip)
                   .concatMap(new Function<FeedsModel, ObservableSource<List<Feed>>>() {
                       @Override
                       public ObservableSource<List<Feed>> apply(@NonNull final FeedsModel feedsModel) throws Exception {
                           feedsModelResult = feedsModel;
                           return Observable.fromCallable(new Callable<List<Feed>>() {
                               @Override
                               public List<Feed> call() throws Exception {

                                   return feedsModel.getFeed();
                               }
                           });
                       }
                   })
                   .concatMap(new Function<List<Feed>, ObservableSource<Feed>>() {
                       @Override
                       public ObservableSource<Feed> apply(@NonNull List<Feed> feeds) throws Exception {

                           return Observable.fromIterable(feeds);
                       }
                   }).filter(new Predicate<Feed>() {
                       @Override
                       public boolean test(@NonNull Feed feed) throws Exception {
                           return feed.getProducts() != null;
                       }
                   })
                   .concatMap(new Function<Feed, ObservableSource<Double>>() {
                       @Override
                       public ObservableSource<Double> apply(@NonNull Feed feed) throws Exception {
                           return Observable.fromIterable((ArrayList<Double>) feed.getProducts());
                       }
                   })
                   .concatMap(new Function<Double, ObservableSource<ProductModel>>() {
                       @Override
                       public ObservableSource<ProductModel> apply(@NonNull Double productId) throws Exception {
                           return productsRepository.fetchProduct(productId.intValue(), shopId, langId, currency);
                       }
                   }).concatMap(new Function<ProductModel, ObservableSource<Map<Integer, ProductModel>>>() {
                       @Override
                       public ObservableSource apply(@NonNull ProductModel productModel) throws Exception {

                           productMap.put(productModel.getIdProduct(), productModel);
                           return Observable.fromCallable(new Callable<Map<Integer, ProductModel>>() {
                               @Override
                               public Map<Integer, ProductModel> call() throws Exception {
                                   return productMap;
                               }
                           });
                       }
                   }).concatMap(new Function<Map<Integer, ProductModel>, ObservableSource<Pair>>() {
                       @Override
                       public ObservableSource apply(@NonNull final Map<Integer, ProductModel> productModelMap) throws Exception {
                           return Observable.fromCallable(new Callable() {
                               @Override
                               public Object call() throws Exception {
                                   return Pair.create(feedsModelResult, productMap);
                               }
                           });
                       }
                   });
       }

更新： 在 onSubscribe 中，我保留对一次性用品的引用，并在获得第一个结果后将其丢弃在 onNext() 中。这是一种有效的方法吗？

在调用的最后结果 Pair.create(feedsModelResult, productMap);我想我应该使用 zipWith 运算符来等待所有结果完成，但我不确定如何

Answer 1

理解您的流程并不容易，但您似乎正在尝试查询一些 FeedsModel 对象，并发出单个 Pair 值，该值将这个 FeedsModel 对象与一些收集的内部产品图一起打包它包含的对象。

问题是您的Observable 可能会被每个内部列表展平，并尝试在到达地图的过程中收集它们，而流仍将这些项目发送给最终订阅者。

您需要的是一个获取输入FeedsModel 的流，将所有项目收集到一个地图并只发出一个作为结果地图的项目，然后您可以将此地图与输入 FeedsModel 以及您刚刚收集的结果地图。
假设您的 feedRepository.fetchFeeds 只能返回单个 FeedsModel 项目（您可以使用 Single<FeedsModel> 对其进行细化），您将在结束流中得到单个结果。

建议使用包含结果选择器的 flatMap 变体：

feedRepository.fetchFeeds(shopId, langId, skip)
 .flatMap(feedsModel ->
         getProductsMapFromFeedsModelObservable(feedsModel, shopId, langId, currency)
         ,(feedsModel, productsMap) ->
               Pair.create(feedsModel, productsMap)
         );

getProductsMapFromFeedsModelObservable() 是一个 Observable，它从输入 FeedsModel 收集产品到产品映射：

private Observable<HashMap<Integer, ProductModel>> getProductsMapFromFeedsModelObservable(
            FeedsModel feedsModel, int shopId, int langId, int currency) {
        return Observable.fromIterable(feedsModel.getFeed())
                .filter(feed -> feed.getProducts() != null)
                .flatMapIterable(feed -> feed.getProducts())
                .flatMap(productId -> productsRepository.fetchProduct(productId.intValue(), shopId, langId, currency))
                .reduce(new HashMap<Integer, ProductModel>(),
                        (productsMap, productModel) -> {
                            productsMap.put(productModel.getIdProduct(), productModel);
                            return productsMap;
                        })
                .toObservable();
    }

reduce() 用于收集所有项目以映射和发出单个项目，flatMap() 用于此处，因为您可以从并行执行中获得，concat() 可能无论如何都不需要，不知道为什么它首先出现放置（保留订单？）。