Spring Integration handleReactive单向处理报错：通道头缺失求助

代码片段 #

@Bean
public IntegrationFlow mqttInFlowTopicConnect(JsonToObjectTransformer jsonToAlarmTransformer){
    return IntegrationFlow.from("alarmChannel")
            .transform(jsonToAlarmTransformer)
            .route("payload.alarmType", m -> m
                    .channelMapping("NewAlarm","newAlarmChannel")
                    .channelMapping("ChangedAlarm","changedAlarmChannel")
                    .channelMapping("ClearedAlarm","clearedAlarmChannel")
            )
            .get();
}

@Bean
public IntegrationFlow newAlarmFlow(){
    return IntegrationFlow.from("newAlarmChannel")
            .handleReactive(message -> Mono.just(message)
                    .flatMap(msg -> alarmRepo.save((Alarm)msg.getPayload()))
                    .then(),
                    e -> e.requiresReply(false)
            );
}

错误信息 #

接收NewAlarm消息时抛出异常：

"no output-channel or replyChannel header available"

错误堆栈：

org.springframework.messaging.MessageHandlingException: error occurred in message handler [ServiceActivator for [org.springframework.integration.handler.MethodInvokingMessageProcessor@788185c7] (handleNewAlarm.insertNewAlarm.serviceActivator)]
at org.springframework.integration.support.utils.IntegrationUtils.wrapInHandlingExceptionIfNecessary(IntegrationUtils.java:191) ~[spring-integration-core-6.3.4.jar:6.3.4]
at org.springframework.integration.handler.AbstractMessageHandler.doHandleMessage(AbstractMessageHandler.java:108) ~[spring-integration-core-6.3.4.jar:6.3.4]
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:73) ~[spring-integration-core-6.3.4.jar:6.3.4]
at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:160) ~[spring-integration-core-6.3.4.jar:6.3.4]
at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:121) ~[spring-integration-core-6.3.4.jar:6.3.4]
at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:72) ~[spring-integration-core-6.3.4.jar:6.3.4]
at org.springframework.integration.channel.AbstractMessageChannel.sendInternal(AbstractMessageChannel.java:390) ~[spring-integration-core-6.3.4.jar:6.3.4]
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:334) ~[spring-integration-core-6.3.4.jar:6.3.4]
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:187) ~[spring-messaging-6.1.13.jar:6.1.13]
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:166) ~[spring-messaging-6.1.13.jar:6.1.13]
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47) ~[spring-messaging-6.1.13.jar:6.1.13]
at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:109) ~[spring-messaging-6.1.13.jar:6.1.13]
at org.springframework.integration.router.AbstractMessageRouter.doSend(AbstractMessageRouter.java:238) ~[spring-integration-core-6.3.4.jar:6.3.4]
at org.springframework.integration.router.AbstractMessageRouter.handleMessageInternal(AbstractMessageRouter.java:220) ~[spring-integration-core-6.3.4.jar:6.3.4]
at org.springframework.integration.handler.AbstractMessageHandler.doHandleMessage(AbstractMessageHandler.java:105) ~[spring-integration-core-6.3.4.jar:6.3.4]
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:73) ~[spring-integration-core-6.3.4.jar:6.3.4]
at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:132) ~[spring-integration-core-6.3.4.jar:6.3.4]
at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:148) ~[spring-integration-core-6.3.4.jar:6.3.4]
at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:121) ~[spring-integration-core-6.3.4.jar:6.3.4]
at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:72) ~[spring-integration-core-6.3.4.jar:6.3.4]
at org.springframework.integration.channel.AbstractMessageChannel.sendInternal(AbstractMessageChannel.java:390) ~[spring-integration-core-6.3.4.jar:6.3.4]
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:334) ~[spring-integration-core-6.3.4.jar:6.3.4]
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:187) ~[spring-messaging-6.1.13.jar:6.1.13]
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:166) ~[spring-messaging-6.1.13.jar:6.1.13]
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47) ~[spring-messaging-6.1.13.jar:6.1.13]
at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:109) ~[spring-messaging-6.1.13.jar:6.1.13]
at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutput(AbstractMessageProducingHandler.java:536) ~[spring-integration-core-6.3.4.jar:6.3.4]
at org.springframework.integration.handler.AbstractMessageProducingHandler.doProduceOutput(AbstractMessageProducingHandler.java:359) ~[spring-integration-core-6.3.4.jar:6.3.4]
at org.springframework.integration.handler.AbstractMessageProducingHandler.produceOutput(AbstractMessageProducingHandler.java:288) ~[spring-integration-core-6.3.4.jar:6.3.4]
at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutputs(AbstractMessageProducingHandler.java:252) ~[spring-integration-core-6.3.4.jar:6.3.4]
at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:151) ~[spring-integration-core-6.3.4.jar:6.3.4]
at org.springframework.integration.handler.AbstractMessageHandler.doHandleMessage(AbstractMessageHandler.java:105) ~[spring-integration-core-6.3.4.jar:6.3.4]
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:73) ~[spring-integration-core-6.3.4.jar:6.3.4]
at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:132) ~[spring-integration-core-6.3.4.jar:6.3.4]
at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:148) ~[spring-integration-core-6.3.4.jar:6.3.4]
at org.springframework.integration.dispatcher.UnicastingDispatcher$1.run(UnicastingDispatcher.java:129) ~[spring-integration-core-6.3.4.jar:6.3.4]
at org.springframework.integration.util.ErrorHandlingTaskExecutor.lambda$execute$0(ErrorHandlingTaskExecutor.java:56) ~[spring-integration-core-6.3.4.jar:6.3.4]
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144) ~[na:na]
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642) ~[na:na]
at java.base/java.lang.Thread.run(Thread.java:1583) ~[na:na]
Caused by: org.springframework.messaging.core.DestinationResolutionException: no output-channel or replyChannel header available
at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutput(AbstractMessageProducingHandler.java:531) ~[spring-integration-core-6.3.4.jar:6.3.4]
at org.springframework.integration.handler.AbstractMessageProducingHandler.doProduceOutput(AbstractMessageProducingHandler.java:359) ~[spring-integration-core-6.3.4.jar:6.3.4]
at org.springframework.integration.handler.AbstractMessageProducingHandler.produceOutput(AbstractMessageProducingHandler.java:288) ~[spring-integration-core-6.3.4.jar:6.3.4]
at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutputs(AbstractMessageProducingHandler.java:252) ~[spring-integration-core-6.3.4.jar:6.3.4]
at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:151) ~[spring-integration-core-6.3.4.jar:6.3.4]
at org.springframework.integration.handler.AbstractMessageHandler.doHandleMessage(AbstractMessageHandler.java:105) ~[spring-integration-core-6.3.4.jar:6.3.4]
... 38 common frames omitted

问题原因 #

虽然设置了requiresReply(false)，但handleReactive返回的Mono<Void>仍会被框架识别为需要发送输出的信号。由于newAlarmChannel是订阅型通道，未携带replyChannel头信息，框架尝试发送输出时找不到目标通道，从而抛出异常。

解决方案 #

方案一：配置handler不发送输出 #

在handleReactive的配置中添加outputChannel(null)，明确告诉框架不需要发送任何输出：

@Bean
public IntegrationFlow newAlarmFlow(){
    return IntegrationFlow.from("newAlarmChannel")
            .handleReactive(message -> Mono.just(message)
                    .flatMap(msg -> alarmRepo.save((Alarm)msg.getPayload()))
                    .then(),
                    e -> e.requiresReply(false).outputChannel(null)
            );
}

方案二：返回空Mono避免产生输出 #

修改Reactive流的返回值为Mono.empty()，让框架认为没有输出需要发送：

@Bean
public IntegrationFlow newAlarmFlow(){
    return IntegrationFlow.from("newAlarmChannel")
            .handleReactive(message -> 
                alarmRepo.save((Alarm) message.getPayload())
                        .then(Mono.empty()),
                e -> e.requiresReply(false)
            );
}

方案三：使用普通handle替代handleReactive（可选） #

如果不需要Reactive特性，也可以改用普通的handle方法，明确不返回任何结果：

@Bean
public IntegrationFlow newAlarmFlow(){
    return IntegrationFlow.from("newAlarmChannel")
            .handle((message, headers) -> {
                alarmRepo.save((Alarm) message.getPayload());
                return null;
            }, e -> e.requiresReply(false));
}

内容的提问来源于stack exchange，提问作者Mario Tuan