使用RxJava和Retrofit进行N次顺序api调用

Question

我有一个文件列表，我想从Android设备上传到后端。由于内存限制，我希望只在第一个API调用完成后进行第二个API调用，在第二个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。创造，你可能做错了……这是一个合理的方法吗?是否有更好的方法与Retrofit的RxJava集成来实现这一点?

Answer 1

我很天真地会这么做(但这行不通，见下文):

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

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

然而，

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)));

Answer 2

RxJava的原生程序将在Observable.from(...)中发出所有项，就像并行地一样。这是考虑平行发射的最好方法。然而，有些情况需要真正的后续执行整个链。我想到了下面的解决方案，可能不是最好的，但有效。

import rx.Observable;
import rx.Subscriber;
import java.util.Iterator;
import java.util.function.Function;
public class Rx {
    public static void ignore(Object arg) {
    }
    public static <E, R> Observable<Void> sequential(Iterator<E> iterator, Function<E, Observable<R>> action) {
        return Observable.create(collectorSubscriber ->
                Observable.<Void>create(producerSubscriber ->
                        producerSubscriber.setProducer(ignoredCount -> {
                            if (!iterator.hasNext()) {
                                producerSubscriber.onCompleted();
                                return;
                            }
                            E model = iterator.next();
                            action.apply(model)
                                    .subscribe(
                                            Rx::ignore,
                                            producerSubscriber::onError,
                                            () -> producerSubscriber.onNext(null));
                        }))
                        .subscribe(new Subscriber<Void>() {
                            @Override
                            public void onStart() {
                                request(1);
                            }
                            @Override
                            public void onCompleted() {
                                collectorSubscriber.onNext(null);
                                collectorSubscriber.onCompleted();
                            }
                            @Override
                            public void onError(Throwable e) {
                                collectorSubscriber.onError(e);
                            }
                            @Override
                            public void onNext(Void aVoid) {
                                request(1);
                            }
                        }));
    }
}

示例用法如下:

    Iterator<? extends Model> iterator = models.iterator();
    Rx.sequential(iterator, model -> someFunctionReturnsObservable(model))
            .subscribe(...);

这个方法保证了

的链式执行

Observable<Dummy> someFunctionReturnsObservable(Model model)

Answer 3

目前创建可观察对象的首选方式是使用fromAsync:

Observable.fromAsync(new Action1<AsyncEmitter<Object>>()
    {
        @Override
        public void call(final AsyncEmitter<Object> emitter)
        {
            emitter.onNext(object);
            emitter.onCompleted();
            emitter.setCancellation(new AsyncEmitter.Cancellable()
            {
                @Override
                public void cancel() throws Exception
                {
                    // on unSubscribe() callback
                }
            });
        }
    }, AsyncEmitter.BackpressureMode.BUFFER);