使用Java DSL/启动作业的Spring批处理集成

Question

我有一个正在工作的Spring Boot/批处理项目，包含2个作业。

我现在尝试添加集成，仅使用java配置/java DSL从远程SFTP轮询文件，然后启动作业。

文件轮询工作正常，但我不知道如何在我的流中启动作业，尽管阅读了以下链接：

一些代码段：

@Bean
public SessionFactory SftpSessionFactory()
    {
        DefaultSftpSessionFactory sftpSessionFactory = new DefaultSftpSessionFactory();
        sftpSessionFactory.setHost("myip");
        sftpSessionFactory.setPort(22);
        sftpSessionFactory.setUser("user");
        sftpSessionFactory.setPrivateKey(new FileSystemResource("path to my key"));

    return sftpSessionFactory;
}

@Bean
public IntegrationFlow ftpInboundFlow() {
    return IntegrationFlows
        .from(Sftp.inboundAdapter(SftpSessionFactory())
            .deleteRemoteFiles(Boolean.FALSE)
            .preserveTimestamp(Boolean.TRUE)
            .autoCreateLocalDirectory(Boolean.TRUE)
            .remoteDirectory("remote dir")
            .regexFilter(".*\\.txt$")
            .localDirectory(new File("C:/sftp/")),
                e -> e.id("sftpInboundAdapter").poller(Pollers.fixedRate(600000)))
        .handle("FileMessageToJobRequest","toRequest")
        // what to put next to process the jobRequest ?

对于.handle（“FileMessageToJobRequest”,“ToRequest”）,我使用的是http://docs.spring.io/spring-batch/trunk/reference/html/springbatchintegration.html

我将非常感谢任何帮助，非常感谢。

在我添加了Gary comment之后编辑，它不会编译--当然--因为我不理解请求是如何传播的：

.handle("FileMessageToJobRequest","toRequest")
.handle(jobLaunchingGw())
.get();
}

@Bean
public MessageHandler jobLaunchingGw() {
    return new JobLaunchingGateway(jobLauncher());
}

@Autowired
private JobLauncher jobLauncher;

@Bean
public JobExecution jobLauncher(JobLaunchRequest req) throws JobExecutionException {
    JobExecution execution = jobLauncher.run(req.getJob(), req.getJobParameters());
    return execution;
}

我找到了一种使用@ServiceActivator启动作业并将其添加到我的流中的方法，但我不确定这是否是一种好的做法：

 .handle("lauchBatchService", "launch")

@Component("lauchBatchService")
public class LaunchBatchService {
    private static Logger log = LoggerFactory.getLogger(LaunchBatchService.class);

@Autowired
private JobLauncher jobLauncher;

@ServiceActivator
public JobExecution launch(JobLaunchRequest req) throws JobExecutionException {

    JobExecution execution = jobLauncher.run(req.getJob(), req.getJobParameters());

    return execution;
}

}

Answer 1

    .handle(jobLaunchingGw())
    // handle result

...

@Bean
public MessageHandler jobLaunchingGw() {
    return new JobLaunchingGateway(jobLauncher());
}

其中joblauncher()是joblauncherbean。

编辑

您的服务激活器的操作与JLG的操作大致相同；它使用这个代码。

@Bean
public JobExecution jobLauncher(JobLaunchRequest req) throws JobExecutionException {
    JobExecution execution = jobLauncher.run(req.getJob(), req.getJobParameters());
    return execution;
}

@Autowired
private JobLauncher jobLauncher;

@Bean
public MessageHandler jobLaunchingGw() {
    return new JobLaunchingGateway(jobLauncher);   
}