Spring集成路由无法发送消息

Question

所以，问题在于路由器。当路由器尝试向通道发送消息时，我会收到错误：Dispatcher没有通道“newTypingNotificationHandler”的订户。输入'。但我有这个频道名称的集成流防御。

@Bean
public IntegrationFlow routeEcmIncomeRq(AbstractMessageRouter typeNotificationRouter) {
    return IntegrationFlows.from(FROM_CHANNEL)
            .routeToRecipients(r -> r
                    .recipientFlow(p -> p instanceof TypingNotificationDto,
                            f -> f.route(typeNotificationRouter)
                    )
                    .defaultOutputChannel(DEFAULT_SERVICE_CHANNEL)
            ).get();
}

@Bean
public AbstractMessageRouter typeNotificationRouter(IncomeRepository incomeRepository) {
    return new AbstractMessageRouter() {
        @Override
        protected Collection<MessageChannel> determineTargetChannels(Message<?> message) {
            TypingNotificationDto messagePayload = (TypingNotificationDto) message.getPayload();

            if (!incomeRepository.existsById(StringUtil.ecdFileIdToUuid(messagePayload.getEcdDocumentImage().getDocumentSource()))) {
                return Collections.singleton(MessageChannels.direct("newTypingNotificationHandler.input").get());
            } else {
                return Collections.singleton(MessageChannels.direct("existsTypingNotificationHandler.input").get());
            }
        }
    };
}

@Bean
public IntegrationFlow newTypingNotificationHandler() {
    return f -> f.log("need's create new Income");
}

@Bean
public IntegrationFlow existsTypingNotificationHandler() {
    return f -> f.log("exist income process");
}

原因：

org.springframework.integration.MessageDispatching异常：调度器在org.springframework.integration.dispatcher.UnicastingDispatcher.do调度（UnicastingDispatcher.java:139）没有订户~[Spring-集成-core-5.2.6. RELEASE. jar： 5.2.6. RELEASE]在org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch（UnicastingDispatcher.java:106）~[Spring-集成-core-5.2.6. RELEASE. jar： 5.2.6. RELEASE]在org.springframework.integration.channel.AbstractSubcribableChannel.do发送（AbstractSubcribableChannel.java:73）~[Spring-集成-core-5.2.6. RELEASE. jar： 5.2.6. RELEASE]在org.springframework.integration.channel.AbstractMessageChannel.send（AbstractMessageChannel.java:461）~[Spring-集成-core-5.2.6. RELEASE. jar： 5.2.6.发布]在org.springframework.integration.channel.AbstractMessageChannel.send（AbstractMessageChannel.java:403）~[sping-整合-core-5.2.6. RELEASE. jar： 5.2.6. RELEASE]在org.springframework.messaging.core.GenericMessagingTemplate.do发送（GenericMessagingTemplate.java:187）~[sping-Message-5.2.6. RELEASE. jar： 5.2.6. RELEASE]在org.springframework.messaging.core.GenericMessagingTemplate. do发送（GenericMessagingTemplate. java： 166）~[sping-Message-5.2.6. RELEASE. jar： 5.2.6. RELEASE]在org. springframework. core. GenericMessagingTemplate-5.2.6. RELEASE. jar： 5.2.6. RELEASE]在org. springframework. mail. core. AbstractMessageSendingTemplate.发送（AbstractMessageSendingTemplate. java： 109）~[sping-Message-5.2.6.

Answer 1

您每次都返回一个新频道，而不是Spring管理的频道

MessageChannels.direct("newTypingNotificationHandler.input").get();

使用

return Collections.singleton(getChannelResolver().resolveDestination("newTypingNotificationHandler.input"));

相反。但是，最好在解析后缓存，而不是为每个消息返回一个新的集合。