发出的每个列表项的 RxJava 延迟

Question

我正在努力实现一些我认为在 Rx 中会相当简单的东西。

我有一个项目列表，我想延迟发出每个项目。

似乎 Rx delay() 运算符只是将所有项目的发射移动了指定的延迟，而不是每个单独的项目。

这是一些测试代码。它将列表中的项目分组。然后每个组在被发射之前应该有一个延迟。

 Observable.range(1, 5)
    .groupBy(n -> n % 5)
    .flatMap(g -> g.toList())
    .delay(50, TimeUnit.MILLISECONDS)
    .doOnNext(item -> {
        System.out.println(System.currentTimeMillis() - timeNow);
        System.out.println(item);
        System.out.println(" ");
    }).toList().toBlocking().first();

结果是：

 154ms
[5]

155ms
[2]

155ms
[1]

155ms
[3]

155ms
[4]

但我希望看到的是这样的：

 174ms
[5]

230ms
[2]

285ms
[1]

345ms
[3]

399ms
[4]

我究竟做错了什么？

原文由 athor 发布，翻译遵循 CC BY-SA 4.0 许可协议

Answer 1

一种方法是使用 zip 将您的 observable 与 Interval observable 结合起来以延迟输出。

 Observable.zip(Observable.range(1, 5)
        .groupBy(n -> n % 5)
        .flatMap(g -> g.toList()),
    Observable.interval(50, TimeUnit.MILLISECONDS),
    (obs, timer) -> obs)
    .doOnNext(item -> {
      System.out.println(System.currentTimeMillis() - timeNow);
      System.out.println(item);
      System.out.println(" ");
    }).toList().toBlocking().first();

原文由 iagreen 发布，翻译遵循 CC BY-SA 3.0 许可协议

Answer 2

最简单的方法似乎就是使用 concatMap 并将每个项目包装在延迟的 Obserable 中。

 long startTime = System.currentTimeMillis();
Observable.range(1, 5)
        .concatMap(i-> Observable.just(i).delay(50, TimeUnit.MILLISECONDS))
        .doOnNext(i-> System.out.println(
                "Item: " + i + ", Time: " + (System.currentTimeMillis() - startTime) +"ms"))
        .toCompletable().await();

印刷：

 Item: 1, Time: 51ms
Item: 2, Time: 101ms
Item: 3, Time: 151ms
Item: 4, Time: 202ms
Item: 5, Time: 252ms

原文由 Magnus 发布，翻译遵循 CC BY-SA 4.0 许可协议