如何访问Spring集成流中间的通量？

Question

我尝试在Spring集成中访问flux对象，而不将流声明拆分为两个函数。我想知道如何执行以下操作：

@Bean
public IntegrationFlow mainFlow() {
   return IntegrationFlows.from(somePublisher)
      // Access publisher here to perform something like:
      .handle(flux -> flux.buffer(Duration.ofMillis(200))
      .handle(writeToS3)
      .get();
}

我不介意将我在评论中提到的通量操作转移到另一个类（可能是为了作为某种网关），但对我来说，从同一个mainFlow函数启动和流显然非常重要，因此它将非常清晰易读，能够理解我在应用程序中所做的事情。我看到了Monos网关的文档，但示例代码甚至不可能（它们讨论的是函数中没有的通量，作为初学者，我很难理解那里发生了什么）。

Answer 1

IntegrationFlows.from（某个Publisher）为提供的Publisher启动一个ractive流。流的其余部分针对源Publisher中的每一个事件执行。因此，您的。句柄（通量-

如果您的想法是为源Publisher应用缓冲区（）并继续下去，那么请考虑使用reactive（）自定义程序：https://docs.spring.io/spring-integration/docs/current/reference/html/reactive-streams.html#fluxmessagechannel-and-reactivestreamsconsumer.

因此，我将使用以下方法代替handle（）：

.bridge(e -> e.reactive(flux -> flux.buffer(Duration.ofMillis(200)))

下一个句柄（WriteToS3）将针对缓冲的List结果执行，因此要小心在WriteToS3中执行和期望的操作。