反应器3. x-限制组的时间By Flux

Question

有没有办法强迫groupBy（）生成的流量在一段时间后完成（或者类似地，限制“开放”组的最大值），而不管上游的完整性？我有以下内容：

Flux<Foo> someFastPublisher;

someFastPublisher
  .groupBy(f -> f.getKey())
  .delayElements(Duration.ofSeconds(1)) // rate limit each group
  .flatMap(g -> g) // unwind the group
  .subscribe()
;

我遇到了通量挂起的情况，假设是因为组的数量大于平面映射的并发性。我可以增加flatMap的并发性，但是没有简单的方法来判断最大可能大小是多少。相反，我知道Foo是按Foo分组的。key将在时间/发布顺序上彼此接近，并且宁愿在groupBy Flux与flatMap并发上使用某种时间窗口（并且最终与两个不同的组一起使用相同的key（），这没什么大不了的）。

我猜想，在一些快速发布者完成之前，通量不会完成，即通量交给平面地图（flatMap），只是保持“开放”（尽管它们不太可能获得新事件）。

我可以通过预取整数来解决这个问题。groupBy或整数中的最大值。最大化并发性-但是有方法控制组的“生命”吗？

Answer 1

是：您可以将获取（持续时间）应用于组，以确保它们提前关闭，并且在那之后将打开具有相同密钥的新组：

source.groupBy(v -> v.intValue() % 2)
      .flatMap(group -> group
              .take(Duration.ofMillis(1000))
              .count()
              .map(c -> "group " + group.key() + " size = " + c)
      )
      .log()
      .blockLast();