在可观察项 RxJava 之间添加延迟 [重复]答案

【问题标题】：Adding delay between Observable Items RxJava [duplicate]在可观察项 RxJava 之间添加延迟 [重复]
【发布时间】：2017-01-10 01:57:00
【问题描述】：

我有一个从对象列表创建的可观察对象。对于列表中的每个对象，我都会发出一个网络请求，但我想在列表中的每个项目之间放置一个延迟，以便稍微隔开请求。这是我的代码的 sn-p。

return Observable.from(documentGroupModels).flatMap(new Func1<DocumentGroupModel, Observable<Boolean>>() {
        @Override
        public Observable<Boolean> call(DocumentGroupModel documentGroupModel) {
            return refreshDocumentWithUri(documentGroupModel.getUri(), documentGroupModel.sectionGroupId,
                                          includeExceptions, false);
        }
    });

据我所知，在这种情况下使用延迟或缓冲区并不完全有效。

【问题讨论】：

标签： java rx-java observable

【解决方案1】：

使用“延迟”运算符，例如

return Observable.from(documentGroupModels).flatMap(new Func1<DocumentGroupModel, Observable<Boolean>>() {
        @Override
        public Observable<Boolean> call(DocumentGroupModel documentGroupModel) {
            return refreshDocumentWithUri(documentGroupModel.getUri(), documentGroupModel.sectionGroupId,
                                          includeExceptions, false).delay(2000, TimeUnit.MILLISECONDS);
        }
    });

【讨论】：

这不会延迟每个项目，它只是将所有项目推进 2 秒。
奇怪。它与第二种解决方案基本相同，只是它使用了您自己的大部分代码和 flatMap（concatMap 更好）。查看接受的答案here。

【解决方案2】：

如果您的延迟是静态的，您可以结合使用 Zip 和 interval 运算符，这样您就可以在每次配置间隔时发出一个 zip 项。

查看示例

       @Test
public void delaySteps() {
    long start = System.currentTimeMillis();
    Subscription subscription =
            Observable.zip(Observable.from(Arrays.asList(1, 2, 3)), Observable.interval(200, TimeUnit.MILLISECONDS),
                           (i, t) -> i)
                    .subscribe(n -> System.out.println("time:" + (System.currentTimeMillis() - start)));
    new TestSubscriber((Observer) subscription).awaitTerminalEvent(3000, TimeUnit.MILLISECONDS);
}

你也可以用你的列表创建一个 Observable 并使用 concatMap，然后你可以为每个发射的项目使用 delay。也许这个解决方案更优雅，没有那么 Hacky

      @Test
public void delayObservableList() {
    Observable.from(Arrays.asList(1, 2, 3, 4, 5))
            .concatMap(s -> Observable.just(s).delay(100, TimeUnit.MILLISECONDS))
            .subscribe(n -> System.out.println(n + " emitted"),
                       e -> {
                       },
                       () -> System.out.println("All emitted"));
    new TestSubscriber().awaitTerminalEvent(1000, TimeUnit.MILLISECONDS);

}

您可以在此处查看另一个延迟示例https://github.com/politrons/reactive/blob/master/src/test/java/rx/observables/utils/ObservableDelay.java

【讨论】：

我选择了第二种解决方案，效果很好。谢谢！