GroupedFlux 中的 Reactor delayElements 延迟所有组中的元素

Question

我有一个用例，我想通过 PartitionKey 创建一堆 GroupedFlux，并在每个组中延迟 100 毫秒。但是，我希望多个组同时启动。因此，如果有 3 个组，我预计每 100 毫秒发出 3 条消息。但是，使用以下代码我每 100 毫秒只看到 1 条消息。

这是我期望运行的代码。

final Flux<GroupedFlux<String, TData>> groupedFlux =
        flux.groupBy(Event::getPartitionKey);
groupedFlux.subscribe(g -> g.delayElements(Duration.ofMillis(100))
        .flatMap(this::doWork)
        .doOnError(throwable -> log.error("error: ", throwable))
        .onErrorResume(e -> Mono.empty())
        .subscribe());

这是日志。

21:24:29.318   parallel-5]  : GroupByKey : 2
21:24:29.424   parallel-6]  : GroupByKey : 3
21:24:29.529   parallel-7]  : GroupByKey : 1
21:24:29.634   parallel-8]  : GroupByKey : 2
21:24:29.739   parallel-9]  : GroupByKey : 3
21:24:29.844  parallel-10]  : GroupByKey : 1
21:24:29.953  parallel-11]  : GroupByKey : 2
21:24:30.059  parallel-12]  : GroupByKey : 3
21:24:30.167   parallel-1]  : GroupByKey : 1

（查看每条日志语句之间几乎有 100 毫秒的差异。1s 列是时间戳。

Answer 1

经过更多分析，我发现它工作正常。我的测试有不正确的 PartitionKey 数据，导致 single GroupedFlux.

回答我自己的问题，以防有人怀疑 delayElements 在 groupedFlux 上的工作方式不同。它没有。

GroupedFlux 中的 Reactor delayElements 延迟所有组中的元素

Reactor delayElements within a GroupedFlux delays elements across all groups

project-reactor