android - 如何在使用 Rx 开始第二个请求之前等待第一个请求完成？

Question

我有一个带有回调的异步方法 makeRequest()。它从我的应用程序的不同类调用了很多次。我需要这个调用一个接一个地开始，而不是同时开始。

我想使用 Rx 来实现它。像这样:

public void execute() { // This method called many times from another classes
    Observable.just(true)
        // what I need to add here?
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .map(o -> {
                    internalExecute();
                    return o;
            })
        .subscribe();
}

private void internalExecute() { // This method should called only when previous call was finished
    makeRequest(this::onRequestFinished);
}

private void onRequestFinished() {
    // here is I handle request finish
}

但是现在所有的请求都是并行的。我需要在这里添加什么才能一个一个地运行请求？

Answer 1

根据评论，您在这里分离了流和请求。每个执行请求的客户端都期望请求的结果。但不允许请求并行运行，在这种情况下，我认为最简单的方法是将 Scheduler 限制为应用程序全局后台顺序线程执行器，即:

Schedulers.from(Executors.newSingleThreadExecutor())

在你的应用中的某处提供这个单线程执行器，当然是以单例方式，每个请求流都使用相同的对象是很重要的:

private final Scheduler singleThreadScheduler = Schedulers.from(Executors.newSingleThreadExecutor());

public void execute() { // This method called many times from another classes
    Observable.just(true)
            .map(o -> {
                internalExecute();
                return o;
            })
            .subscribeOn(singleThreadScheduler)
            .subscribe();
}

private void internalExecute() { // This method should called only when previous call was finished
    makeRequest(this::onRequestFinished);
}

private void onRequestFinished() {
    //NOTE: you should make sure that the callback execute where you need it (main thread?)

    // here is I handle request finish
}

除此之外，您不会将 Observable 暴露给客户端，而是使用回调机制，您可以通过使 execute() 进一步利用响应式(Reactive)方法返回 Observable。 (并享受 Obesrvable 的组合、运算符、正确使用 observeOn/subscribeOn、使用 onError 进行错误处理，处理/取消订阅等)，当你使用异步 api 时，你可以使用 fromEmitter()/create()(在较新的 RxJava1 版本中))，阅读更多 here :

 private final Scheduler singleThreadScheduler = Schedulers.from(Executors.newSingleThreadExecutor());

public Observable<Result> execute() { // This method called many times from another classes
    return Observable.fromEmitter(new Action1<Emitter<? extends Object>>() {
        @Override
        public void call(Emitter<?> emitter) {
            emitter.setCancellation(() -> {
                //cancel request on unsubscribing
            });
            makeRequest(result -> {
                emitter.onNext(result);
            });
        }
    })
            .subscribeOn(singleThreadScheduler)
}