列表中的每个项目的RxJava在Android中使用Retofit发出api请求

Question

我的Android项目中存在以下问题：

我有一个使用Retrofit从上一个网络调用获得的元素列表（ArrayList）。CategoryItem看起来像这样：

public class CategoryItem {
   String id;
   String name;
   public CategoryItem(String id, String name) {
     this.id = id;
     this.name = name;
   }
   public String getId() {
     return id;
   }
   public String getName() {
     return name;
   }
}

实际上，类别由20个元素组成。现在，我需要制作一个网络API，并获取每个类别的产品列表。为此，产品API将类别中的ID作为查询参数。在获得某种类别的产品列表之后，我将其添加到内部SQLite DB中。

我想使用RxJava（1和2）做到这一点。

我到目前为止所做的不是多线程的，而是在MainThread上的，这是不正确的。我将在此处添加代码段：

for (int i = 0; i < categoryItems.size(); i++) {
    CategoryItem categoryItem = categoryItems.get(i);
    final String iCat = categoryItem.getId();
    Observable<ProductResponse> call = networkManager.getApiServiceProductsRx().getProductsRx(iCat);

    Subscription subscription = call
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Subscriber<ProductResponse>() {
                @Override
                public void onCompleted() {
                }

                @Override
                public void onError(Throwable e) {
                    if (e instanceof HttpException) {
                        HttpException response = (HttpException) e;
                        int code = response.code();
                        Log.d("RetrofitTest", "Error code: " + code);
                    }
                }

                @Override
                public void onNext(ProductResponse response) {
                    mProductItemsBelongingToACategory = new ArrayList<>();
                    mProductItemsBelongingToACategory = response.getProductChannel().getProductItems();
                    mDatabaseHandler.addProducts(iCat, mProductItemsBelongingToACategory);
                }
            });

    subscriptions2.add(subscription);

}

我想用flatMap或flatMapIterable做到这一点，并最终使用多线程。

Answer 1

RxJava具有等效的for循环：

Observable.from(...) //RxJava1

要么

Observable.fromIterable(...) //RxJava2

应用它可以使您遍历CategoryItem列表。然后，每个都categoryItem.getId()必须与getProductsRx()API调用的结果配对。然后将所有结果合并到单个列表中，toList()如下所示：

RxJava1：

Observable.from(categoryItems)
    .flatMap(new Func1<CategoryItem, Observable<Pair<String, ProductResponse>>>() {
        @Override
        public Observable<Pair<String, ProductResponse>> call(@NonNull CategoryItem categoryItem) throws Exception {
            return Observable.zip(
                    Observable.just(categoryItem. getId()),
                    networkManager.getApiServiceProductsRx().getProductsRx(categoryItem. getId()),
                    new Func2<String, ProductResponse, Pair<String, ProductResponse>>() {
                        @Override
                        public Pair<String, ProductResponse> call(@NonNull String id, ProductResponse productResponse) throws Exception {
                            return new Pair<String, ProductResponse>(id, productResponse);
                        }
                    });
        }
    })
   .toList()
   .subscribeOn(Schedulers.io())
   .observeOn(AndroidSchedulers.mainThread())

RxJava2：

Observable.fromIterable(categoryItems)
    .flatMap(new Function<CategoryItem, ObservableSource<Pair<String, ProductResponse>>>() {
        @Override
        public ObservableSource<Pair<String, ProductResponse>> apply(@NonNull CategoryItem categoryItem) throws Exception {
            return Observable.zip(
                    Observable.just(categoryItem. getId()),
                    networkManager.getApiServiceProductsRx().getProductsRx(categoryItem. getId()),
                    new BiFunction<String, ProductResponse, Pair<String, ProductResponse>>() {
                        @Override
                        public Pair<String, ProductResponse> apply(@NonNull String id, @NonNull ProductResponse productResponse) throws Exception {
                            return new Pair<String, ProductResponse>(id, productResponse);
                        }
                    });
        }
    })
   .toList()
   .subscribeOn(Schedulers.io())
   .observeOn(AndroidSchedulers.mainThread())

用RxJava2编写。

以上仅Observable是s。您仍然可以使用Subscriber并将数据保存到中的数据库onNext()。但是，您也可以doOnNext()在之后使用运算符toList()将数据保存到数据库中。