java - 发出的每个列表项的 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]

我做错了什么？

Answer 1

执行此操作的最简单方法似乎是使用 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