使用Rx Java Observable一次进行多个异步调用（触发并忘记调用）

jack 发表于 Dev

插口

我有需要异步异步调用的下游api调用列表（约10个）。直到现在我一直在使用可调用对象

List<RequestContextPreservingCallable <FutureResponse>> callables

我会将api调用添加到此列表中，并在最后使用executeAsyncNoReturnRequestContextPreservingCallables提交。

使用Rx Java Observables我该怎么做？

List<RequestContextPreservingCallable<FutureResponse>> callables = new 
ArrayList<RequestContextPreservingCallable<FutureResponse>>();

callables.add(apiOneConnector.CallToApiOne(name));
callables.add(apiTwoConnector.CallToApiTWO(sessionId));
....

//execute all the calls
executeAsyncNoReturnRequestContextPreservingCallables(callables);

萨拉斯·肯（Sarath Kn）

您可以利用zip运算符。该zip运营商可以采取多种观测，并同时执行它们，所有的结果已经抵达后，将继续进行。

然后，您可以将这些结果转换为所需的形式并传递到下一个级别。

根据您的示例。假设您有多个用于获取名称和会话等的API调用，如下所示

Observable.zip(getNameRequest(), getSessionIdRequest(), new BiFunction<String, String, Object>() {
        @Override
        public Object apply(String name, String sessionId) throws Exception {
            // here you will get all the results once everything is completed. you can then take these 
            // results and transform into another object and returnm from here. I decided to transform the results into an Object[]
            // the retuen type of this apply funtion is generic, so you can choose what to return
            return new Object[]{name, sessionId};
        }
    })
    .subscribeOn(Schedulers.io())  // will start this entire chain in an IO thread
    .observeOn(AndroidSchedulers.mainThread()) // observeOn will filp the thread to the given one , so that the downstream will be executed in the specified thread. here I'm switching to main at this point onwards
    .subscribeWith(new DisposableObserver<Object>() {
        @Override
        public void onNext(Object finalResult) {
           // here you will get the final result with all the api results
        }

        @Override
        public void onError(Throwable e) {
            // any error during the entire process will be triggered here
        }

        @Override
        public void onComplete() {
             //will be called once the whole chain is completed and terminated
        }
    });

您甚至可以将观察值列表传递给zip，如下所示

    List<Observable<String>> requests = new ArrayList<>();
    requests.add(getNameRequest());
    requests.add(getSessionIdRequest());

    Observable.zip(requests, new Function<Object[], Object[]>() {
        @Override
        public Object[] apply(Object[] objects) throws Exception {
            return new Object[]{objects[0], objects[1]};
        }
    }).subscribeWith(new DisposableObserver<Object[]>() {
        @Override
        public void onNext(Object[] objects) {

        }

        @Override
        public void onError(Throwable e) {

        }

        @Override
        public void onComplete() {

        }
    })

本文收集自互联网，转载请注明来源。

如有侵权，请联系 [email protected] 删除。