RxJava - 如何使用 doOnNext 等待异步任务的结果

Question

如何实现 doOnNext 等待多个异步任务的结果？

例如-

public void getImages(User user) {
    Flowable.create(new FlowableOnSubscribe<User>() {
        @Override
        public void subscribe(@io.reactivex.rxjava3.annotations.NonNull FlowableEmitter<User> emitter) throws Throwable {
            emitter.onNext(user);
        }
    }, BackpressureStrategy.BUFFER)
            .observeOn(Schedulers.io())
            .doOnNext(user -> {
                ArrayList<String> imagesUrls = user.getUrls();
                for (String url : imagesUrls) {
                    storage.getReference().child("images").child(url).getBytes(ParametersConventions.FIREBASE_DOWNLOAD_IMAGE_MAX_SIZE).
                    addOnSuccessListener(bytes -> {
                       doSomething(bytes);    
                    });
                }
            })
            .doOnNext(user -> {
                doSomething();
            })
            .doOnComplete(...);
}

我希望在所有下载图像的异步调用完成后调用 doSomething 的 doOnNext。

Answer 1

将该 API 调用转换为响应式类型并将其合并到主流程中：

int max = ParametersConventions.FIREBASE_DOWNLOAD_IMAGE_MAX_SIZE;

public Completable downloadAsync(URL url) {
    return Completable.create(inner -> {
                           storage.getReference()
                           .child("images")
                           .child(url)
                           .getBytes(max)
                           .addOnSuccessListener(bytes -> {
                                doSomething(bytes);
                                inner.onComplete();
                           });
                     });
}

一起：

Flowable.create(emitter-> {
       emitter.onNext(user);
    }, BackpressureStrategy.BUFFER)
            .observeOn(Schedulers.io())
            .concatMapSingle(user -> 
                 Flowable.fromIterable(user.getUrls())
                     .concatMapCompletable(url -> downloadAsync(url))
                     .andThen(Single.just(user))
             )
             .doOnNext(user -> {
                 doSomething();
             })
             .doOnComplete(...);

Answer 2

doOnNext 运算符会在每次流中有新项目时触发，因此它不是您的最佳选择。根据您的需要尝试使用 map/flatMap/concatMap 运算符。如果您需要打几个电话，然后对数据做一些事情，您可以查看我已经回答过的类似问题链接：Chaining API Requests with Retrofit + Rx 您可以在其中找到一种方法来进行顺序网络调用，然后对数据列表执行任何操作：D