我有一個需要異步調(diào)用的下游 api 調(diào)用列表（大約 10 個）。到目前為止，我一直在使用 callablesList<RequestContextPreservingCallable <FutureResponse>> callables我會將 api 調(diào)用添加到此列表中，并在最后使用 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 callsexecuteAsyncNoReturnRequestContextPreservingCallables(callables);
1 回答
慕神8447489
TA貢獻(xiàn)1780條經(jīng)驗 獲得超1個贊
您可以使用zip運算符。該zip運營商可以采取多種觀測，并同時執(zhí)行它們，所有的結(jié)果已經(jīng)抵達(dá)后，將繼續(xù)進(jìn)行。
然后，您可以將這些結(jié)果轉(zhuǎn)換為您需要的形式并傳遞到下一個級別。
按照你的例子。假設(shè)您有多個 API 調(diào)用來獲取名稱和會話等，如下所示
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
}
});
您甚至可以將 observables 列表傳遞給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() {
}
})
添加回答
舉報
0/150
提交
取消