撰写多个网络调用RxJava - Android

Question

帮助编写多个网络调用并在Rxjava中累积结果。 （我在Android应用程序中使用。）

State 
 -- List<City> cityList;

 City
 - cityId;

RestCall 1
Observable<State> stateRequest = restService.getStates();

RestCall 2
Observable<CityDetail> cityRequest = restService.getCityDetail(cityId);

在UI中，我必须在获取每个城市的所有详细信息后显示城市列表，然后在列表视图中显示。如何实现parllel网络调用并累积结果。 ？

我希望将所有城市细节结果放入源状态“对象”中的列表中。由于状态对象有一些需要消除的信息。这可能吗？

stateRequest ??? 
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Subscriber<State>() {
    @Override
    public void onCompleted() {

    }

    @Override
    public void onError(Throwable e) {
    }

    @Override
    public void onNext(State result) {
    // Get city list and display
    }
});

我查看了这个示例，它显示了我们如何能够更多地压缩一个可观察的响应。下面的代码片段显示了3个可观察量。但在我的情况下，我必须并行或顺序进行20次网络调用（我的意思是在后台但是一个接一个）。我如何实现这一目标。任何帮助或指示？

https://gist.github.com/skehlet/9418379

Observable.zip(f3Observable, f4Observable, f5Observable, new Func3<String, Integer, Integer, Map<String, String>>() {
    @Override
    public Map<String, String> call(String s, Integer integer, Integer integer2) {
        Map<String, String> map = new HashMap<String, String>();
        map.put("f3", s);
        map.put("f4", String.valueOf(integer));
        map.put("f5", String.valueOf(integer2));
        return map;
    }

Answer 1

我认为您的代码可以简化为类似的东西，因为您对zip运算符的使用接近使用toList运算符

 stateRequest
 .subscribe(State state ->  {
     Observable.from(state.getCityList())
               .flatMap(City city -> restService.getCityDetail(city.getId())
               .toList()
               .subscribe(List<City> cities -> {

                     state.clear();
                     state.addAll(cities);
               });
     });

由于RxJava不提供throttle运算符，您可以构建类似这样的东西：

Observable<City> limiter = Observable.zip(Observable.interval(1, SECONDS), aCity, (i, c) -> c);

使用这个，限制器是一个可观察的，每秒会发出一个城市。

因此，使用您的代码，如果您想限制对getCityDetail的调用，例如：

 Observable<Object> limiter = Observable.interval(1, SECONDS);
 stateRequest
 .subscribe(State state ->  {
     Observable.zip(limiter, Observable.from(state.getCityList()), (i, c) -> c)
               .flatMap(City city -> restService.getCityDetail(city.getId())
               .toList()
               .subscribe(List<City> cities -> {

                     state.clear();
                     state.addAll(cities);
               });
     });

Answer 2

stateRequest
.flatMap(new Func1<State, Observable<State>>() {
    @Override
    public Observable<State> call(final State state) {

        List<Observable> cityObservablesList = new ArrayList<Observable>();

        for(City city: state.getCityList()) {
            cityObservablesList.add(restService.getCityDetail(city.getId());
        }

        Observable cityObservables = Observable.from(cityObservablesList);
        return Observables.zip(cityObservables, new FuncN<State>() {
            @Override
            public State call(Object... args) {
                List<City> cityList = state.getCityList();
                cityList.clear();
                for(Object object: args) {
                    cityList.add((City)object);
                }

                return state;
            }
        })
    })
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Subscriber<State>() {
    @Override
    public void onCompleted() {

    }

    @Override
    public void onError(Throwable e) {
    }

    @Override
    public void onNext(State result) {
    // Get city list and display
    }
});

我使用zip操作符和Iterable for city list作为第一个参数。但我面临另一个问题。由于zip并行执行作业，因此并行执行10-15个网络调用，并且服务器拒绝每秒最大查询错误（QPS-403）。我如何指示zip操作员一个接一个地执行任务？

我确实通过向city observable添加延迟[delay（c * 200，TimeUnit.MILLISECONDS））]来解决这个问题。但似乎不是一个合适的解决方案。

有什么建议？

Answer 3

看看flatMap（Function ..，BiFunction ..）。也许那就是你需要的。

statesRepository.getStates()
   .flatMap(states-> Observable.fromIterable(states))
   .flatMap(
           state-> cityRepository.getStateCities(state),
           (state, cityList) -> {
               state.setCities(cityList);
               return state;
           })
   .subscribe(state-> showStateWithCity(state));