反应堆核心，取30然后等待再取30答案

【问题标题】：Reactor core, take 30 then wait before taking another 30反应堆核心，取30然后等待再取30
【发布时间】：2021-10-30 02:19:43
【问题描述】：

使用reactor.core.publisher.Flux，如何从Flux 中获取n 个值，然后等待一段时间再获取下一批？

onboardService
  .loadRepositories(user)               // Flux of values
  .take(30)                             // Take 30 from the flux
  .delayElements(Duration.ofMinutes(1)) // Wait one minute
  .doOnEach(...)                        // Process the value
  .???                                  // How to repeat with the next 30?

在链的进一步，我将每个值从 Flux 推送到每分钟速率限制为 30 的服务。

onboardService
  .loadRepositories(user)   
  .limitRate(10)
  .delayElements(Duration.ofSeconds(10))

听起来像我想要的，但它的行为并不符合我的预期。将它与这些参数一起使用，它在处理每个单独的通量之间等待 10 秒，而我希望它处理 10 秒，然后再花 10 秒。

有没有更好的方法不让终端服务超载？

【问题讨论】：

标签： java spring-webflux project-reactor

【解决方案1】：

您可以将通量窗口化并将每个窗口限制为一分钟：

onboardService.loadRepositories(user)
  .transform(flux ->
               flux
                 .window(30)
                 .zipWith(Flux.interval(Duration.ZERO, Duration.ofMinutes(1)))
                 .flatMap(Tuple2::getT1))
  .doOnEach(...) // Process the value

【讨论】：

我不知道.window 这太棒了！