android - 在 Android 中使用 RxJava 对任务进行排队

Question

我正在开发具有后台数据同步功能的 Android 应用程序。我目前正在使用 RxJava 定期在服务器上发布一些数据。除此之外，我想为用户提供一个“强制同步”按钮，它将立即触发同步。我知道如何使用 Observable.interval() 以固定的时间间隔推送数据，并且我知道如何使用 Observable.just() 来推送强制的数据，但如果发生在前一个仍在运行时触发的情况，我想将它们排队。

因此，让我们以 1 分钟为自动同步间隔为例，假设同步持续 40 秒(为了更容易理解，我在这里夸大了)。现在，如果有任何机会，用户在自动仍在运行时按下“强制”按钮(反之亦然 - 当强制仍在运行时自动触发)，我想将第二个同步请求排队等待第一个结束。

我已经画了这张图片，这可能会给它带来更多的视角:

如您所见，自动触发(通过一些 Observable.interval())，在同步过程中，用户按下“强制”按钮。现在我们要等待第一个请求完成，然后重新开始强制请求。在某一时刻，当强制请求运行时，新的自动请求再次被触发，刚刚将其添加到队列中。在队列中的最后一个完成后，一切都停止了，然后稍后再次安排自动。

希望有人能指出我正确的运算符(operator)如何做到这一点。我已尝试使用 Observable.combineLatest()，但队列列表在开始时已分派(dispatch)，当我将新同步添加到队列时，在前一个操作完成时它没有继续。

非常感谢任何帮助，达科

Answer 1

可以通过合并定时器和按钮点击Observable/Subject，利用onBackpressureBuffer的排队效果和concatMap处理进入它，确保一次运行一个。

PublishSubject<Long> subject = PublishSubject.create();

Observable<Long> periodic = Observable.interval(1, 1, TimeUnit.SECONDS);

periodic.mergeWith(subject)
.onBackpressureBuffer()
.concatMap(new Func1<Long, Observable<Integer>>() {
    @Override
    public Observable<Integer> call(Long v) {
        // simulates the task to run
        return Observable.just(1)
                .delay(300, TimeUnit.MILLISECONDS);
    }
}
).subscribe(System.out::println, Throwable::printStackTrace);

Thread.sleep(1100);
// user clicks a button
subject.onNext(-1L);

Thread.sleep(800);