RxJava 为列表中的每个项目在 Android 中使用 Retrofit 发出 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 上，这是不正确的。 我将在这里添加代码sn-p：

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 编写。

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