问题 RxJava组合请求序列

该 flatMap 运营商旨在满足这些类型的工作流程。

我将通过一个简单的五步示例来概述广泛的笔画。希望您可以轻松地在代码中重建相同的原则：

@Test fun flatMapExample() {
    // (1) constructing a fake stream that emits a list of values
    Observable.just(listOf(1, 2, 3, 4, 5))
            // (2) convert our List emission into a stream of its constituent values 
            .flatMap { numbers -> Observable.fromIterable(numbers) }
            // (3) subsequently convert each individual value emission into an Observable of some 
            //     newly calculated type
            .flatMap { number ->
                when(number) {
                       1 -> Observable.just("A1")
                       2 -> Observable.just("B2")
                       3 -> Observable.just("C3")
                       4 -> Observable.just("D4")
                       5 -> Observable.just("E5")
                    else -> throw RuntimeException("Unexpected value for number [$number]")
                }
            }
            // (4) collect all the final emissions into a list
            .toList()
            .subscribeBy(
                    onSuccess = {
                        // (5) handle all the combined results (in list form) here
                        println("## onNext($it)")
                    },
                    onError = { error ->
                        println("## onError(${error.message})")
                    }
            )
}

（顺便说一下，如果排放的顺序很重要，请看看使用情况 concatMap 代替）。

我希望有所帮助。

检查下面它是否正常工作。

假设你有多个网络调用，你需要make-cals来获取Github用户信息和Github用户事件。

并且您希望在更新UI之前等待每个返回。 RxJava可以在这里帮助你。 让我们首先定义我们的Retrofit对象来访问Github的API，然后为两个网络请求调用设置两个observable。

Retrofit repo = new Retrofit.Builder()
        .baseUrl("https://api.github.com")
        .addConverterFactory(GsonConverterFactory.create())
        .addCallAdapterFactory(RxJavaCallAdapterFactory.create())
        .build();

Observable<JsonObject> userObservable = repo
        .create(GitHubUser.class)
        .getUser(loginName)
        .subscribeOn(Schedulers.newThread())
        .observeOn(AndroidSchedulers.mainThread());

Observable<JsonArray> eventsObservable = repo
        .create(GitHubEvents.class)
        .listEvents(loginName)
        .subscribeOn(Schedulers.newThread())
        .observeOn(AndroidSchedulers.mainThread());

用于它的接口如下：

public interface GitHubUser {
  @GET("users/{user}")
  Observable<JsonObject> getUser(@Path("user") String user);
}

public interface GitHubEvents {
  @GET("users/{user}/events")
  Observable<JsonArray> listEvents(@Path("user") String user);
}

我们使用RxJava之后 压缩 结合我们的两个Observable并在创建一个新的Observable之前等待它们完成的方法。

Observable<UserAndEvents> combined = Observable.zip(userObservable, eventsObservable, new Func2<JsonObject, JsonArray, UserAndEvents>() {
  @Override
  public UserAndEvents call(JsonObject jsonObject, JsonArray jsonElements) {
    return new UserAndEvents(jsonObject, jsonElements);
  }
});

最后，让我们在新的组合Observable上调用subscribe方法：

combined.subscribe(new Subscriber<UserAndEvents>() {
          ...
          @Override
          public void onNext(UserAndEvents o) {
            // You can access the results of the 
            // two observabes via the POJO now
          }
        });

没有更多的线程等待网络调用完成。 RxJava已经在zip（）中为您完成了所有这些。 希望我的回答可以帮到你。

我用RxJava2解决了类似的问题。并行执行Api 2请求会略微加快工作速度。

private InformationRepository informationRepository;

//init....

public Single<List<FullInformation>> getFullInformation() {
    return informationRepository.getInformationList()
            .subscribeOn(Schedulers.io())//I usually write subscribeOn() in the repository, here - for clarity
            .flatMapObservable(Observable::fromIterable)
            .flatMapSingle(this::getFullInformation)
            .collect(ArrayList::new, List::add);

}

private Single<FullInformation> getFullInformation(Information information) {
    return informationRepository.getExtendedInformation(information)
            .map(extendedInformation -> new FullInformation(information, extendedInformation))
            .subscribeOn(Schedulers.io());//execute requests in parallel
}

InformationRepository - 只是界面。它的实现对我们来说并不有趣。

public interface InformationRepository {

    Single<List<Information>> getInformationList();

    Single<ExtendedInformation> getExtendedInformation(Information information);
}

FullInformation - 结果的容器。

public class FullInformation {

    private Information information;
    private ExtendedInformation extendedInformation;

    public FullInformation(Information information, ExtendedInformation extendedInformation) {
        this.information = information;
        this.extendedInformation = extendedInformation;
    }
}

尝试使用 Observable.zip() 运营商。它将等待两个Api呼叫完成后再继续流。然后你可以通过调用插入一些逻辑 flatMap() 之后。

http://reactivex.io/documentation/operators/zip.html