使用RxJava和Retrofit进行N次连续的api调用

Question

我有一个文件列表，我想从Android设备上传到后端。由于内存的限制，我只想在第一个完成后进行第二个API调用，第二个完成后进行第三个API调用，依此类推。

我写了类似

private Observable<Integer> uploadFiles(List<File> files) {
        return Observable.create(subscriber -> {
            for (int i = 0, size = files.size(); i < size; i++) {
                UploadModel uploadModel = new UploadModel(files.get(0));
                int uploadResult = retrofitApi.uploadSynchronously(uploadModel);
                subscriber.onNext(uploadResult);
            }
            subscriber.onCompleted();
        }).subscribeOn(Schedulers.newThread());
    }

但是我觉得这可能与Rx的精神背道而驰，俗话说，如果您使用的是Observable.create，则可能做错了……这是一种合理的方法吗？通过Retrofit的RxJava集成，是否有更好的方法来实现此目的？

Answer 1

天真地，我会这样做（但是，它不起作用，请参见下文）：

return Observable.from(files).concatMap(file -> retrofitApi.upload(uploadModel));

现在的问题是，没有办法告诉改造只使用一个线程进行这些调用。

reduce但是，会将一个函数调用的结果与原始可观察到的下一个发射值一起传递给下一个。那可以工作，但是传递给函数reduce必须是同步的。不好。

另一种方法是递归地修改可观察对象：

void getNextFile(int i) {
    return retrofit.upload(i).
        onNext(result -> getNextFile(i + 1));
}

大致。但是我不确定如何清理它以使其更具可读性。

我认为最干净的是：

Observable.from(files).map(file -> retrofitApi.uploadSynchronously(new UploadModel(file)));