在 RXJava 中实现异步 API 调用序列

Question

我有一组从本地数据库中检索到的对象，需要按 API 调用的顺序上传到服务器。 对于每个本地数据库对象，我必须随后调用两个异步 API 调用（methodA()、methodB()）。完成整个循环后，我需要调用另一个 API 调用，如下所示。

for(Object object: localDBObjects){
    methodA() -> methodB()
}
methodC()

我的问题是如何阻止 methodC() 调用直到完成循环。

Answer 1

这个怎么样：

public class Main {

public static CountDownLatch finishLatch = new CountDownLatch(1); 

public static Integer methodA(Integer obj) {
    try {
        Thread.sleep((int) (Math.random() * 1000)); //Simulate asynchronous call
    } catch (InterruptedException e) {
    }
    System.out.println("methodA for " + obj + " executed by " + Thread.currentThread().getName());
    return obj;
}

public static Integer methodB(Integer obj) {
    try {
        Thread.sleep((int) (Math.random() * 1000)); //Simulate asynchronous call
    } catch (InterruptedException e) {
    }
    System.out.println("methodB for " + obj + " executed by " + Thread.currentThread().getName());
    return obj;
}

public static void methodC() {
    System.out.println("methodC executed by " + Thread.currentThread().getName());
    finishLatch.countDown(); //Allow main to finish
}

public static void main(String[] args) throws IOException, InterruptedException {
    List<Integer> objectsFromDb = Arrays.asList(1, 2, 3, 4, 5); //List of objects from the DB

    Observable.from(objectsFromDb) 
            .flatMap(obj -> Observable.fromCallable(() -> methodA(obj)).subscribeOn(Schedulers.io())) //Asynchronously call method A
            .flatMap(obj -> Observable.fromCallable(() -> methodB(obj)).subscribeOn(Schedulers.io())) //Asynchronously call method B
            .doOnCompleted(() -> methodC()) //When finished, call methodC
            .subscribe();

    finishLatch.await(); //Wait for everything to finish
}

}

样本输出：

methodA for 5 executed by RxCachedThreadScheduler-5
methodA for 2 executed by RxCachedThreadScheduler-2
methodA for 1 executed by RxCachedThreadScheduler-1
methodB for 1 executed by RxCachedThreadScheduler-2
methodB for 2 executed by RxCachedThreadScheduler-5
methodB for 5 executed by RxCachedThreadScheduler-6
methodA for 3 executed by RxCachedThreadScheduler-3
methodA for 4 executed by RxCachedThreadScheduler-4
methodB for 3 executed by RxCachedThreadScheduler-1
methodB for 4 executed by RxCachedThreadScheduler-2
methodC executed by RxCachedThreadScheduler-2

Answer 2

由于我没有很多关于你的项目和方法的具体实现的信息，即它们的参数和返回类型，我有两个假设。

注意：如果我使用 lambda 表达式，希望您不介意。

1)。方法返回 Observable<Object> 就像 Retrofit
在这种情况下，它们看起来像这样：

public Observable<Object> methodA(Object o){
    return null;
}

public Observable<Object> methodB(Object o){
    return null;
}

public Observable<Object> methodC(Object[] objects){
    return null;
}

对于这种情况，您可以使用以下内容：

Object[] localDBObjects = new Object[10];
Observable.just(localDBObjects)
        .flatMap(objects -> Observable.from(objects)
                                .flatMap(object -> methodA(object))
                                .flatMap(resultFromMethodA -> methodB(resultFromMethodA))
                                .toList())
        .flatMap(listOfResultsFromMethodB -> methodC(listOfResultsFromMethodB.toArray(new Object[listOfResultsFromMethodB.size()])))
        .subscribe(resultFromMethodC -> {
            //do something
        }, t -> t.printStackTrace());

2)。在其他情况下，方法返回 Object，如下所示：

public Object methodA(Object o){
    return null;
}

public Object methodB(Object o){
    return null;
}

public Object methodC(Object[] objects){
    return null;
}

在这种情况下，您需要在某些地方将运算符 flatMap( ) 更改为 map( )：

    Object[] localDBObjects = new Object[10];
    Observable.just(localDBObjects)
            .flatMap(objects -> Observable.from(objects)
                                    .map(object -> methodA(object))
                                    .map(resultFromMethodA -> methodB(resultFromMethodA))
                                    .toList())
            .map(listOfResultsFromMethodB -> methodC(listOfResultsFromMethodB.toArray(new Object[listOfResultsFromMethodB.size()])))
            .subscribe(resultFromMethodC -> {
                //do something
            }, t -> t.printStackTrace());