reactive() 端点

从 5.5 版本开始，ConsumerEndpointSpec 提供了一个 reactive() 配置属性，并带有一个可选的自定义程序 Function<? super Flux<Message<?>>, ? extends Publisher<Message<?>>>。此选项将目标端点配置为 ReactiveStreamsConsumer 实例，独立于输入通道类型，通过 IntegrationReactiveUtils.messageChannelToFlux() 将其转换为 Flux。提供的函数用于 Flux.transform() 运算符来自定义（publishOn()、log()、doOnNext() 等）来自输入通道的反应式流源。

以下示例演示如何独立于最终订阅者和生产者更改来自输入通道的发布线程到该 DirectChannel

@Bean
public IntegrationFlow reactiveEndpointFlow() {
    return IntegrationFlow
            .from("inputChannel")
            .transformWith(t -> t
                              .<String, Integer>transformer(Integer::parseInt)
                              .reactive(flux -> flux.publishOn(Schedulers.parallel()))
            )
            .get();
}

有关更多信息，请参阅 反应式流支持。