spring - Spring Integration 在聚合后调用另一个处理程序方法

Question

我正在开发一个系统，它将从目录中读取和处理文件。处理完所有文件后，它将调用一个方法，该方法又生成一个文件。此外，它应该根据文件名路由/处理文件，我也使用了 spring 集成路由器。以下是集成的代码片段。.channel(aggregatorOutputChannel())我的问题是，如果我删除任何一行或，这将不起作用.channel(confirmChannel())，而且我必须.channel(aggregatorOutputChannel())在聚合器之前和之后保持相同的频道。为什么我需要所有 3 通道声明？如果这是错误的如何纠正它。

我正在使用 JDK 8、Spring 5、Spring boot 2.0.4。

@Configuration
@EnableIntegration
public class IntegrationConfig {

    @Value("${agent.demographic.input.directory}")
    private String inputDir;

    @Value("${agent.demographic.output.directory}")
    private String outputDir;

    @Value("${confirmationfile.directory}")
    private String confirmDir;

    @Value("${input.scan.frequency: 2}")
    private long scanFrequency;

    @Value("${processing.waittime: 6000}")
    private long messageGroupWaiting;

    @Value("${thread.corepoolsize: 10}")
    private int corepoolsize;

    @Value("${thread.maxpoolsize: 20}")
    private int maxpoolsize;

    @Value("${thread.queuecapacity: 1000}")
    private int queuedepth;

    @Bean
    public MessageSource<File> inputFileSource() {
        FileReadingMessageSource src = new FileReadingMessageSource();

        src.setDirectory(new File(inputDir));
        src.setAutoCreateDirectory(true);

        ChainFileListFilter<File> chainFileListFilter = new ChainFileListFilter<>();
        chainFileListFilter.addFilter(new AcceptOnceFileListFilter<>() );
        chainFileListFilter.addFilter(new RegexPatternFileListFilter("(?i)^.+\\.xml$"));
        src.setFilter(chainFileListFilter);
        return src;
    }

    @Bean
    public UnZipTransformer unZipTransformer() {
        UnZipTransformer unZipTransformer = new UnZipTransformer();
        unZipTransformer.setExpectSingleResult(false);
        unZipTransformer.setZipResultType(ZipResultType.FILE);
        unZipTransformer.setDeleteFiles(true);

        return unZipTransformer;
    }

    @Bean("agentdemographicsplitter")
    public UnZipResultSplitter splitter() {
        UnZipResultSplitter splitter = new UnZipResultSplitter();
        return splitter;
    }

    @Bean
    public DirectChannel outputChannel() {
        return new DirectChannel();
    }

    @Bean
    public DirectChannel aggregatorOutputChannel() {
        return new DirectChannel();
    }

    @Bean("confirmChannel")
    public DirectChannel confirmChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageHandler fileOutboundChannelAdapter() {
        FileWritingMessageHandler adapter = new FileWritingMessageHandler(new File(outputDir));
        adapter.setDeleteSourceFiles(true);
        adapter.setAutoCreateDirectory(true);
        adapter.setExpectReply(true);
        adapter.setLoggingEnabled(true);
        return adapter;
    }


    @Bean
    public MessageHandler confirmationfileOutboundChannelAdapter() {
        FileWritingMessageHandler adapter = new FileWritingMessageHandler(new File(confirmDir));
        adapter.setDeleteSourceFiles(true);
        adapter.setAutoCreateDirectory(true);
        adapter.setExpectReply(false);
        adapter.setFileNameGenerator(defaultFileNameGenerator() );
        return adapter;
    }

    @Bean
    public TaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(corepoolsize);
        executor.setMaxPoolSize(maxpoolsize);
        executor.setQueueCapacity(queuedepth);
        return executor;
    }

    @Bean
    public DefaultFileNameGenerator defaultFileNameGenerator() {
        DefaultFileNameGenerator defaultFileNameGenerator = new DefaultFileNameGenerator();
        defaultFileNameGenerator.setExpression("payload.name");
        return defaultFileNameGenerator;
    }

    @Bean
    public IntegrationFlow confirmGeneration() {
        return IntegrationFlows.
                from("confirmChannel")
                .handle(confirmationfileOutboundChannelAdapter())
                .get();
    }

    @Bean
    public IntegrationFlow individualProcessor() {
        return flow -> flow.handle("thirdpartyIndividualAgentProcessor","processfile").channel(outputChannel()).handle(fileOutboundChannelAdapter());
    }

    @Bean
    public IntegrationFlow firmProcessor() {
        return flow -> flow.handle("thirdpartyFirmAgentProcessor","processfile").channel(outputChannel()).handle(fileOutboundChannelAdapter());
    }

    @Bean
    public IntegrationFlow thirdpartyAgentDemographicFlow() {
        return IntegrationFlows
                .from(inputFileSource(), spec -> spec.poller(Pollers.fixedDelay(scanFrequency,TimeUnit.SECONDS)))
                .channel(MessageChannels.executor(taskExecutor()))
                .<File, Boolean>route(f -> f.getName().contains("individual"), m -> m
                        .subFlowMapping(true, sf -> sf.gateway(individualProcessor()))
                        .subFlowMapping(false, sf -> sf.gateway(firmProcessor()))
                        )
                .channel(aggregatorOutputChannel())
                .aggregate(aggregator -> aggregator.groupTimeout(messageGroupWaiting).correlationStrategy(new CorrelationStrategy() {

                    @Override
                    public Object getCorrelationKey(Message<?> message) {
                        return "xyz";
                    }
                }))
                .channel(aggregatorOutputChannel())
                .handle("agentDemograpicOutput","generateAgentDemographicFile")
                .channel(confirmChannel())
                .get();
    }
}

下面是日志

2018-09-07 17:29:20.003 DEBUG 10060 --- [ taskExecutor-2] o.s.integration.channel.DirectChannel    : preSend on channel 'outputChannel', message: GenericMessage [payload=C:\thirdpartyintg\input\18237232_firm.xml, headers={replyChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@1a867ae7, errorChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@1a867ae7, file_name=18237232_firm.xml, file_originalFile=C:\thirdpartyintg\input\18237232_firm.xml, id=dd70999a-8b8d-93d2-1a43-a961ac2c339f, file_relativePath=18237232_firm.xml, timestamp=1536366560003}]
2018-09-07 17:29:20.003 DEBUG 10060 --- [ taskExecutor-2] o.s.i.file.FileWritingMessageHandler     : fileOutboundChannelAdapter received message: GenericMessage [payload=C:\thirdpartyintg\input\18237232_firm.xml, headers={replyChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@1a867ae7, errorChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@1a867ae7, file_name=18237232_firm.xml, file_originalFile=C:\thirdpartyintg\input\18237232_firm.xml, id=dd70999a-8b8d-93d2-1a43-a961ac2c339f, file_relativePath=18237232_firm.xml, timestamp=1536366560003}]
2018-09-07 17:29:20.006 DEBUG 10060 --- [ taskExecutor-2] o.s.integration.channel.DirectChannel    : postSend (sent=true) on channel 'outputChannel', message: GenericMessage [payload=C:\thirdpartyintg\input\18237232_firm.xml, headers={replyChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@1a867ae7, errorChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@1a867ae7, file_name=18237232_firm.xml, file_originalFile=C:\thirdpartyintg\input\18237232_firm.xml, id=dd70999a-8b8d-93d2-1a43-a961ac2c339f, file_relativePath=18237232_firm.xml, timestamp=1536366560003}]
2018-09-07 17:29:20.006 DEBUG 10060 --- [ taskExecutor-2] o.s.integration.channel.DirectChannel    : postSend (sent=true) on channel 'firmProcessor.input', message: GenericMessage [payload=C:\thirdpartyintg\input\18237232_firm.xml, headers={replyChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@1a867ae7, errorChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@1a867ae7, file_name=18237232_firm.xml, file_originalFile=C:\thirdpartyintg\input\18237232_firm.xml, id=0e6dcb75-db99-1740-7b58-e9b42bfbf603, file_relativePath=18237232_firm.xml, timestamp=1536366559761}]
2018-09-07 17:29:20.007 DEBUG 10060 --- [ taskExecutor-2] o.s.integration.channel.DirectChannel    : preSend on channel 'thirdpartyintgAgentDemographicFlow.channel#2', message: GenericMessage [payload=C:\thirdpartyintg\output\18237232_firm.xml, headers={file_originalFile=C:\thirdpartyintg\input\18237232_firm.xml, id=e6e2a30a-60b9-7cdd-84cc-4977d4c21c97, file_name=18237232_firm.xml, file_relativePath=18237232_firm.xml, timestamp=1536366560007}]
2018-09-07 17:29:20.008 DEBUG 10060 --- [ taskExecutor-2] o.s.integration.channel.DirectChannel    : postSend (sent=true) on channel 'thirdpartyintgAgentDemographicFlow.channel#2', message: GenericMessage [payload=C:\thirdpartyintg\output\18237232_firm.xml, headers={file_originalFile=C:\thirdpartyintg\input\18237232_firm.xml, id=e6e2a30a-60b9-7cdd-84cc-4977d4c21c97, file_name=18237232_firm.xml, file_relativePath=18237232_firm.xml, timestamp=1536366560007}]
2018-09-07 17:29:20.009 DEBUG 10060 --- [ taskExecutor-2] o.s.integration.channel.DirectChannel    : postSend (sent=true) on channel 'thirdpartyintgAgentDemographicFlow.subFlow#1.channel#0', message: GenericMessage [payload=C:\thirdpartyintg\input\18237232_firm.xml, headers={file_originalFile=C:\thirdpartyintg\input\18237232_firm.xml, id=13713de8-91ce-b1fa-f52d-450d3038cf9c, file_name=18237232_firm.xml, file_relativePath=18237232_firm.xml, timestamp=1536366559757}]
2018-09-07 17:29:26.009  INFO 10060 --- [ask-scheduler-9] o.s.i.a.AggregatingMessageHandler        : Expiring MessageGroup with correlationKey[processdate]
2018-09-07 17:29:26.011 DEBUG 10060 --- [ask-scheduler-9] o.s.integration.channel.NullChannel      : message sent to null channel: GenericMessage [payload=C:\thirdpartyintg\output\17019222_individual.xml, headers={file_originalFile=C:\thirdpartyintg\input\17019222_individual.xml, id=c654076b-696f-25d4-bded-0a43d1a8ca97, file_name=17019222_individual.xml, file_relativePath=17019222_individual.xml, timestamp=1536366559927}]
2018-09-07 17:29:26.011 DEBUG 10060 --- [ask-scheduler-9] o.s.integration.channel.NullChannel      : message sent to null channel: GenericMessage [payload=C:\thirdpartyintg\output\18237232_firm.xml, headers={file_originalFile=C:\thirdpartyintg\input\18237232_firm.xml, id=e6e2a30a-60b9-7cdd-84cc-4977d4c21c97, file_name=18237232_firm.xml, file_relativePath=18237232_firm.xml, timestamp=1536366560007}]

标签： springspring-bootspring-integration

Answer 1

首先RegexPatternFileListFilter应该是第一个ChainFileListFilter。这样您就不会占用AcceptOnceFileListFilter您不感兴趣的文件中的内存。

最后你需要.channel(confirmChannel())，thirdpartyAgentDemographicFlow因为这是你confirmGeneration流程的输入。

我认为你根本不需要.channel(aggregatorOutputChannel())隐含。.channel(outputChannel())您在子流程中也不需要它。

这是行不通的

请详细说明：你得到什么错误，它是如何工作的等等......你也可以分享一些调试日志org.springframework.integration来确定你的消息是如何传播的。

如果您在 GitHub 上分享一些简单的 Spring Boot 项目，让我们根据您提供的说明进行操作和重现，也会有很大帮助。

更新

此外，我注意到您的聚合器基于groupTimeout(). 要使其将聚合消息发送到下游，您还需要在此处进行配置：

/**
 * @param sendPartialResultOnExpiry the sendPartialResultOnExpiry.
 * @return the handler spec.
 * @see AbstractCorrelatingMessageHandler#setSendPartialResultOnExpiry(boolean)
 */
public S sendPartialResultOnExpiry(boolean sendPartialResultOnExpiry) {

默认情况false下，因此您的消息确实会发送到NullChannel. 在文档中查看更多信息：https ://docs.spring.io/spring-integration/docs/current/reference/html/messaging-routing-chapter.html#agg-and-group-to