Spring Integration-“当前组件”（routerFlow）是单向的“MessageHandler”，不适合配置“输出通道”

Question

我实现了如下所示的RecipientListRouter-

    @Bean
    public RecipientListRouter routerFlow() {
        RecipientListRouter router = new RecipientListRouter();
        router.setIgnoreSendFailures(true);
        router.setApplySequence(true);
        router.addRecipient("channelChkn", "headers.get('eventSubType').contains('CHKN')");
        router.addRecipient("channelBkd", "headers.get('eventSubType').contains('BKD')");
        router.addRecipient("channelBrd", "headers.get('eventSubType').contains('BRD')");
        router.addRecipient("channelAciRecCncl", "headers.get('eventSubType').contains('ACI-REC-CNCL')");
        router.addRecipient("channelSeatAsgn", "headers.get('eventSubType').contains('SEATNBR-')");
        router.addRecipient("channelDeboard", "headers.get('isDeBoarded') == true");
        router.setDefaultOutputChannelName(IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME);
        LOGGER.info("********************* RecipientListRouter *********************" + router.getRecipients());
        return router;
    }

    @Bean
    public IntegrationFlow baseEventFlow() {

        return IntegrationFlows.from(Kafka.messageDrivenChannelAdapter(kafkaMessageListenerContainer))
                .filter(filterMessage, "rejectPastData")
                .transform(aciMessageTransformer, "parserXMLMessage")
                .route(routerFlow())
                // executor used to parallelise the multiple subscribe execution
                .publishSubscribeChannel(Executors.newCachedThreadPool(),
                        pubsub -> pubsub.subscribe(flow -> flow.channel("pubCountEvntChannel"))
                                .subscribe(flow -> flow.channel("pubTravlerEventChannel")))
                .get();

    }

router.addRecipient("pubSubChannel");

###编辑

    @Bean
    public IntegrationFlow baseEventFlow() {

        return IntegrationFlows.from(Kafka.messageDrivenChannelAdapter(kafkaMessageListenerContainer))
                .filter(filterMessage, "rejectPastData")
                .transform(aciMessageTransformer, "parserXMLMessage")
                .gateway(eventFlow())
                .handle (test())
                .get()
   }

    @Bean
    public IntegrationFlow eventFlow() {
        return f -> f.route(routerFlow());
    }

    @Bean
    public RecipientListRouter routerFlow() {
        RecipientListRouter router = new RecipientListRouter();
        router.setIgnoreSendFailures(true);
        router.setApplySequence(true);
        router.addRecipient("channelChkn", "headers.get('eventSubType').contains('CHKN')");
        router.addRecipient("channelBkd", "headers.get('eventSubType').contains('BKD')");
        router.addRecipient("channelBrd", "headers.get('eventSubType').contains('BRD')");
        router.addRecipient("channelAciRecCncl", "headers.get('eventSubType').contains('ACI-REC-CNCL')");
        router.addRecipient("channelSeatAsgn", "headers.get('eventSubType').contains('SEATNBR-')");
        // router.addRecipient("channelSeatAsgn","headers.get('eventSubType').contains('SEATNBR-ASSIGN')");
        // router.addRecipient("channelSeatAsgn","headers.get('eventSubType').contains('SEATNBR-CHG')");
        router.addRecipient("channelDeboard", "headers.get('isDeBoarded') == true");
        //router.setDefaultOutputChannelName(IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME);
        LOGGER.info("********************* RecipientListRouter *********************" + router.getRecipients());
        return router;
    }

Answer 1

您不能在路由器之后继续流。

如果路由器下游的各种流返回一个结果，则必须使用.gateway(...)元素调用路由器；结果将返回到主流。