java - 使用 RxJava 同步两个异步 API 调用

Question

我们可以通过什么方式使用 RxJava 同步两个异步调用？在下面的示例中，作为 API 调用的方法 contentService.listContents 必须先完成，然后才能为每个架构执行 processSchema 方法。

schemaService.listSchema()
    .toObservable()
    .flatMapIterable(schemas -> {
        schemas.forEach(schema -> {
            // async call
            contentService.listContents(schema.getName()).subscribe(contents -> {
                   doSomethingWithThe(contents); 
            });
        });
        // contentService.listContents` must complete first before 
        // processSchema should be called for each schema
        return schemas;
    }).subscribe(schema -> { processSchema(schema); }, 
                 error -> { Console.error(error.getMessage()); });

processSchema 上面的代码的问题不会等待 contentService.listContents，因为它是异步的，彼此之间不同步。

Answer 1

您必须使用 flatMap 来处理 schemas，并且由于它是一个列表，因此您必须再次展开它并 flatMap:

schemaService.listSchema()
.toObservable()
.flatMap(schemas -> 
     Observable.fromIterable(schemas)
     .flatMap(schema -> 
         contentService.listContents(schema.getName())
         .doOnNext(contents -> doSomethingWith(contents))
     )
     // probably you don't care about the inner contents
     .ignoreElements()
     // andThen will switch to this only when the sequence above completes
     .andThen(Observable.just(schemas))
)
.subscribe(
    schema -> processSchema(schema), 
    error -> Console.error(error.getMessage())
);

请注意，您尚未定义服务调用的返回类型，因此您可能必须使用 flatMapSingle 和 doOnSuccess 等。