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

delay java rx-java

2022-08-31 13:54:47

我正在努力实现一些我认为在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]

我做错了什么？

答案 1

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

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

答案 2

一种方法是使用 zip 将可观察性与间隔可观察性组合在一起以延迟输出。

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