如何用RxJava2并行生成多个非阻塞服务请求

Question

我需要一些使用RxJava2实现并行异步调用的帮助

1） 我有多个保险公司（目前我只接受两个），我需要使用该保险公司的名称发送多个并行请求。

2）如果其中任何一个给服务器错误，那么剩余的请求不应该被阻止。

以下是我迄今为止所做的尝试；

ArrayList<String> arrInsurer = new ArrayList<>();
        arrInsurer.add(AppConstant.HDFC);
        arrInsurer.add(AppConstant.ITGI);

        RequestInterface service = getService(ServiceAPI.CAR_BASE_URL);
        for (String insurerName : arrInsurer) {
            service.viewQuote(Utils.getPrefQuoteId(QuoteListActivity.this), insurerName)
                    .subscribeOn(Schedulers.computation())
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(new Consumer<ViewQuoteResDTO>() {
                        @Override
                        public void accept(@NonNull ViewQuoteResDTO viewQuoteResDTO) throws Exception {
                            Log.e("Demo", viewQuoteResDTO.getPremiumData().getIDV()+"");
                             updateList();
                        }
                    }, new Consumer<Throwable>() {
                        @Override
                        public void accept(@NonNull Throwable throwable) throws Exception {
                            Log.e("Demo", throwable.getMessage());
                        }
                    });
        }

private RequestInterface getService(String baseUrl) {      
    Gson gson = new GsonBuilder()
            .setLenient()
            .create();

    return new Retrofit.Builder()
            .baseUrl(baseUrl)                
           .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
            .addConverterFactory(GsonConverterFactory.create(gson))
            .build().create(RequestInterface.class);

}

现在，只有当两个请求都成功响应时，上述代码才能正常工作。但是当任何请求作为内部服务器错误给出响应时，其余请求也会被阻止。

当任何请求给出失败响应时，我得到以下日志错误；

E/Demo: HTTP 500 Aww Snap, Some thing happened at server. Please try back again later.
E/Demo: unexpected end of stream on Connection{100.xxx.xxx.xx:portNo, proxy=DIRECT@ hostAddress=/100.xxx.xxx.xx:portNo cipherSuite=none protocol=http/1.1}

如何处理此错误？

Answer 1

我想像其他与Rx相关的问题一样，这有多个答案。我会给你我在我们的应用程序中使用的，并准确解决这个用例。希望有帮助。

简短版本-这依赖于mergeDelayError。在这里查看

为什么合并？因为与concat不同，它将并行执行可观察对象。为什么mergeDelayError？它会延迟错误...本质上它会执行每个可观察对象并在一切完成时传递错误。这确保即使一个或多个错误，其他错误仍将被执行。

您必须小心一些细节。事件的顺序不再保留，这意味着合并运算符可能会交错一些可观察的事件（考虑到您以前的做事方式，这应该不是问题）。据我所知，即使多个可观察对象失败，您也只会得到一个onError调用。如果这两个都可以，那么您可以尝试以下操作：

List<Observable<ViewQuoteResDTO>> observables = new ArrayList<>();
for (String insurerName : arrInsurer) {     
   observables.add(service.viewQuote(
         Utils.getPrefQuoteId(QuoteListActivity.this), insurerName));
}

Observable.mergeDelayError(observables)
          .subscribeOn(Schedulers.computation())
          .observeOn(AndroidSchedulers.mainThread())
          .subscribe(/* subscriber calls if you need them */);

其思想是创建您要运行的所有可观察对象，然后使用mergeDelayError触发它们。