ReactorGroupBy：取消GroupeFlux后剩余的项目会发生什么？

Question

我需要按高基数键对无限通量进行分组。

例如：

1. 组密钥是域url
2. 对一个域的调用应该是严格顺序的（下一次调用发生在前一次调用完成后）
3. 对不同域的调用应该是并发的
4. 具有相同键（url）的项目之间的时间间隔未知，但预计具有突发性质。几个项目在短时间内发出，然后长时间暂停，直到下一组。

queue
    .groupBy(keyMapper, groupPrefetch)
    .flatMap(
       { group ->
           group.concatMap(
               { task -> makeSlowRemoteCall(task) },
               0
           )
           .takeUntil { remoteCallResult -> remoteCallResult == DONE }
           .timeout(groupTimeout, Mono.empty())
           .then()
       }
      , concurrency
    )

我在两种情况下取消了小组：

>

makeSlowRemoteCall（）结果表明，在不久的将来，该组中很可能没有新项目。

下一项在组超时期间不发出。我使用timeout（timeout，fallback）变量来抑制TimeoutException，并允许flatMap的内部发布器成功完成。

我希望未来可能的项目具有相同的键，以生成新的GroupedFlux，并使用相同的flatMap内部管道进行处理。

但是当我取消它时，如果GroupeFlux还有未请求的项目会发生什么？

groupBy运算符是否使用相同的密钥将它们重新排队到新组中，或者它们永远丢失。如果以后有什么合适的方法来解决我的问题。在这种情况下，我也不确定是否需要将concatMap（）预取设置为0。

Answer 1

我认为，groupBy（）运算符不适合我的任务，因为它有无限的源代码和大量的组。它产生无限个组，因此有必要以某种方式取消下游的空闲组。但不可能在保证没有未使用元素的情况下取消GroupedFlux。

我认为拥有发射有限群的groupBy变体会很好。类似于groupBy（keyMapper，boundryPredicate）。当boundryPredicate返回true时，当前组已完成，具有相同键的下一个元素将启动新组。