如何使用 Spring 批量集成在 IntegrationFlow 中添加自定义 header？

Question

我创建了一个 pollableChannel，它正在侦听 S3 Bucket 获取文件并启动作业。

我的班级是这样的：

    @Bean
    public S3SessionFactory s3SessionFactory(AmazonS3 pAmazonS3) {
        return new S3SessionFactory(pAmazonS3);
    }

    @Bean
    public S3InboundFileSynchronizer s3InboundFileSynchronizer(S3SessionFactory s3SessionFactory) {
        S3InboundFileSynchronizer synchronizer = new S3InboundFileSynchronizer(s3SessionFactory);
        synchronizer.setPreserveTimestamp(true);
        synchronizer.setDeleteRemoteFiles(false);
        synchronizer.setRemoteDirectory(awsS3Properties.getCercBucket());
        return synchronizer;
    }

    @Bean
    public S3InboundFileSynchronizingMessageSource s3InboundFileSynchronizingMessageSource(
            S3InboundFileSynchronizer s3InboundFileSynchronizer) {
        S3InboundFileSynchronizingMessageSource messageSource = new S3InboundFileSynchronizingMessageSource(
                s3InboundFileSynchronizer);
        messageSource.setAutoCreateLocalDirectory(true);
        messageSource.setLocalDirectory(new FileSystemResource(integrationProperties.getTempDirectoryName()).getFile());
        return messageSource;
    }

    @Bean("${receivable.integration.inChannel}")
    public PollableChannel s3FilesChannel() {
        return new QueueChannel();
    }

    @Bean
    public IntegrationFlow integrationFlow(
            S3InboundFileSynchronizingMessageSource s3InboundFileSynchronizingMessageSource) {
        return IntegrationFlows
                .from(s3InboundFileSynchronizingMessageSource,
                        c -> c.poller(Pollers.fixedRate(1000).maxMessagesPerPoll(1)))
                .transform(fileMessageToJobRequest()).handle(jobLaunchingGateway())
                .get();
    }

    @Bean
    public FileMessageToJobRequest fileMessageToJobRequest() {
        FileMessageToJobRequest fileMessageToJobRequest = new FileMessageToJobRequest();
        fileMessageToJobRequest.setFileParameterName("input.file.name");
        fileMessageToJobRequest.setJob(receivablePositionJob);
        return fileMessageToJobRequest;
    }

    @Bean
    @ServiceActivator(inputChannel = "${receivable.integration.inChannel}", poller = @Poller(fixedRate = "1000"))
    public JobLaunchingGateway jobLaunchingGateway() {
        SimpleJobLauncher simpleJobLauncher = new SimpleJobLauncher();
        simpleJobLauncher.setJobRepository(jobRepository);
        simpleJobLauncher.setTaskExecutor(new SyncTaskExecutor());

        JobLaunchingGateway jobLaunchingGateway = new JobLaunchingGateway(simpleJobLauncher);
        jobLaunchingGateway.setOutputChannel(s3FilesChannel());
        return jobLaunchingGateway;
    }

而我的 FileMessageToJobRequest 是这样的：

public class FileMessageToJobRequest {

    private Job job;

    private String fileParameterName;

    public void setFileParameterName(String fileParameterName) {
        this.fileParameterName = fileParameterName;
    }

    public void setJob(Job job) {
        this.job = job;
    }

    @Transformer
    public JobLaunchRequest toRequest(Message<File> message) {
        JobParametersBuilder jobParametersBuilder = new JobParametersBuilder();

        jobParametersBuilder.addString(fileParameterName, message.getPayload().getAbsolutePath());

        return new JobLaunchRequest(job, jobParametersBuilder.toJobParameters());
    }

}

我想在消息中添加自定义 MessageHeader，或者我的第二个选项是在发布消息之前拦截上下文，因为我需要在 ThreadLocal 中设置我的租户。

我该怎么做？

提前致谢。

使用 enrichHeaders 进行更新：

    @Bean
    public IntegrationFlow integrationFlow(
            S3InboundFileSynchronizingMessageSource s3InboundFileSynchronizingMessageSource) {
        return IntegrationFlows
                .from(s3InboundFileSynchronizingMessageSource,
                        c -> c.poller(Pollers.fixedRate(1000).maxMessagesPerPoll(1)))
                .transform(fileMessageToJobRequest())
                .enrichHeaders(Map.of("teste", "testandio"))
                .handle(jobLaunchingGateway())
                .get();
    }

Answer 1

首先，您必须删除 @ServiceActivator(inputChannel = "${receivable.integration.inChannel}"，因为它指向相同的 s3FilesChannel，它也是 JobLaunchingGateway 的 outputChannel。因此，您正在使用这样的配置进行循环。完全不确定它是如何为您工作的...

要在发送到 JobLaunchingGateway 之前添加 header，您只需在 integrationFlow 定义中的 .handle(jobLaunchingGateway()) 之前添加 enrichHeaders()。

如何使用 Spring 批量集成在 IntegrationFlow 中添加自定义 header？

How to add custom header in IntegrationFlow with Spring Batch Integration?

spring-batch

spring-integration

spring-integration-aws