如何处理反应性弹簧积分中的反应性类型？

Question

我正在使用反应性弹簧集成进行一些播放，并尝试执行以下基本操作：

    @Override
    protected IntegrationFlowDefinition<?> buildFlow() {
        return from(someReactiveInboundChannelAdapter)
                .handle(new ReactiveMessageHandlerAdapter(flux -> flux.subscribe(System.out::println)));
    }

但是这不起作用，因为该框架无法识别flux是flux＆lt;？ ＆gt;实例。该框架将其视为消息＆lt;？＆gt;，我不知道如何以及在哪里可以开始编写反应堆代码。

Answer 1

没错。 reactiveVemessageHandlerAdapter期望eacterivevemessagehandler哪个合同是：

Mono<Void> handleMessage(Message<?> message);

我不确定是什么促使您认为输入必须是flux 。

我假设您的somereActiviNBoundChannelAdapter是MessageProducerSupport的扩展，并且确实sisscribetepublisher（发布者＆lt;？ 。这样做的目的是以反应性的方式从源中获取数据，但仍会产生每个事件作为向下游通道的消息。因此，应该清楚地表明，handle（）将接收一件带有单个项目的消息，而不是整个flux。

如果您想将流程视为通量，请考虑使用fluxtransform（）运算符。

同样，最好不要订阅自己，而是在配置和启动结束时在框架中做到这一点。

从技术上讲，您不应该考虑反应性类型。您只需要分别配置流量并仅为单个项目编写逻辑：框架为您提供反应性交互。项目反应堆是围绕flux和mono的库。这就是我们谈论反应类型的地方。春季集成是一个消息框架，通过消息进行通信。最后，对于每一个消息，端点之间的相互作用是否以反应性方式进行。因此，您的处理逻辑可以没有反应性类型。

udate

如果您希望从该flux完全控制somereActiveInboundChannelAdapter，那么您需要做到这一点：

    @Bean
    public Publisher<Message<Object>> reactiveFlow() {
          return IntegrationFlows.from(someReactiveInboundChannelAdapter)
          .toReactivePublisher();
    }

然后注入该> 出版商只要您需要使用它。然后do flux.from（Publisher）以及您需要的任何反应性操作员，包括subscribe（）。

其中一些样本在这里： https://github.com/spring-projects/spring-spring-spring-integration/spring-integration/blob/main/main/spring-integration-integration-integration-integration-integration-core-core-in-src/src/src/test/java/java/java/java/org/spremppramewrame-framewarmewame一下/integration/dsl/reaectivestreams/reaectivestreamstests.java

IntegrationFlowAdapter不能用于此类配置，因为它不能接受IntegrationFlow作为IntegrationFlow代码> buildflow（）。

但是，fluxtransform（）也可以为您做到这一点：

.fluxTransform(flux -> flux.as(Mono::just))

因此，下游流的有效载荷将是flux＆lt; message＆lt;？您可以应付自己。从该fluxtransform（）返回的mono将由框架订阅。它的价值的通量是您在下游流程中的责任。然后，您可以使用普通MessageHandler：

.handle(m -> ((Flux<Message<?>>) m.getPayload())....subscribe())