从 Spring Integration 启动 Spring Batch Job

Question

我需要从远程 SFTP 服务器下载文件并使用 Spring Batch 处理它们。我已经使用 Spring Integration 实现了代码来下载文件。但我无法从 Spring 集成组件启动 Spring Batch 作业。我有以下代码：

    @Autowired
private JobLauncher jobLauncher;

public String OUTPUT_DIR = "temp_dir";

@Value("${sftp.remote.host}")
private String sftpRemoteHost;

@Value("${sftp.remote.user}")
private String sftpUsername;

@Value("${sftp.remote.password}")
private String sftpPassword;

@Value("${sftp.remote.folder}")
private String sftpFolder;

@Bean
public DefaultSftpSessionFactory sftpSessionFactory() {
    final DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory();
    factory.setHost(sftpRemoteHost);
    factory.setAllowUnknownKeys(true);
    factory.setUser(sftpUsername);
    factory.setPassword(sftpPassword);
    return factory;
}

@Bean
public SftpInboundFileSynchronizer sftpInboundFileSynchronizer() {
    final SftpInboundFileSynchronizer fileSynchronizer = new SftpInboundFileSynchronizer(sftpSessionFactory());
    fileSynchronizer.setDeleteRemoteFiles(false);
    fileSynchronizer.setRemoteDirectory(sftpFolder);
    fileSynchronizer.setFilter(new SftpSimplePatternFileListFilter("*.csv"));
    return fileSynchronizer;
}

@Bean
@InboundChannelAdapter(channel = "sftpChannel", poller = @Poller(fixedDelay = "5000"))
public MessageSource<File> sftpMessageSource() {
    final SftpInboundFileSynchronizingMessageSource source =
            new SftpInboundFileSynchronizingMessageSource(sftpInboundFileSynchronizer());
    source.setLocalDirectory(new File(OUTPUT_DIR));
    source.setAutoCreateLocalDirectory(true);
    source.setLocalFilter(new AcceptOnceFileListFilter<>());
    return source;
}

@Bean
@ServiceActivator(inputChannel = "sftpChannel")
public MessageHandler handler() {
    final FileWritingMessageHandler handler = new FileWritingMessageHandler(new File(OUTPUT_DIR));
    handler.setFileExistsMode(FileExistsMode.REPLACE);
    handler.setExpectReply(true);
    handler.setOutputChannelName("parse-csv-channel");
    return handler;
}

@ServiceActivator(inputChannel = "parse-csv-channel", outputChannel = "job-channel")
public JobLaunchRequest adapt(final File file) throws Exception {
    final JobParameters jobParameters = new JobParametersBuilder().addString(
            "input.file", file.getAbsolutePath()).toJobParameters();
    return new JobLaunchRequest(batchConfiguration.job(), jobParameters);
}

@ServiceActivator(inputChannel = "job-channel", outputChannel = "finish")
public JobLaunchingMessageHandler jobHandler(JobLaunchRequest request) throws JobExecutionException {
    return new JobLaunchingMessageHandler(jobLauncher);//.launch(request);
}

@ServiceActivator(inputChannel = "finish")
public void finish() {
    System.out.println("FINISH");
}

但这不起作用（最后一个方法 adapt 中的错误），因为找不到 File 类型的 bean。我无法将这两个部分放在一起。如何接线集成和批处理？

Answer 1

您肯定只需要从您的adapt() 方法中删除@Bean 注释。如果我们真的构建MessageHandler bean，我们需要@Bean，例如JobLaunchingMessageHandler 来接受JobLaunchRequest 有效负载：https://docs.spring.io/spring-batch/trunk/reference/html/springBatchIntegration.html#launching-batch-jobs-through-messages。

在参考手册中查看有关消息注释的更多信息：https://docs.spring.io/spring-integration/docs/4.3.12.RELEASE/reference/html/configuration.html#annotations_on_beans

更新

@Bean
@ServiceActivator(inputChannel = "sftpChannel")
public MessageHandler handler() {
    final FileWritingMessageHandler handler = new FileWritingMessageHandler(new File(OUTPUT_DIR));
    handler.setFileExistsMode(FileExistsMode.REPLACE);
    handler.setExpectReply(true);
    handler.setOutputChannelName("parse-csv-channel");
    return handler;
}

@ServiceActivator(inputChannel = "parse-csv-channel", outputChannel = "job-channel")
public JobLaunchRequest adapt(final File file) throws Exception {
    final JobParameters jobParameters = new JobParametersBuilder().addString(
            "input.file", file.getAbsolutePath()).toJobParameters();
    return new JobLaunchRequest(batchConfiguration.job(), jobParameters);
}

@Bean
@ServiceActivator(inputChannel = "job-channel")
public JobLaunchingGateway jobHandler() {
    JobLaunchingGateway jobLaunchingGateway = new JobLaunchingGateway(jobLauncher);
    jobLaunchingGateway.setOutputChannelName("finish");
    return jobLaunchingGateway;
}