文件可用后立即运行 Spring Batch Job

Question

我想在共享文件夹中的文件可用时立即运行我的 Spring Batch 作业。我已经做了一个观察服务来检查目录中的新条目，但是我将如何触发我的批处理作业？

我做了以下更改 -

  @Component
public class ScheduleJob {

    private Logger log = LoggerFactory.getLogger(this.getClass());

    @Autowired
    private JobLauncher jobLauncher;

    @Autowired
    private UtilizationBatchConfiguration utilizationBatchConfiguration;

    @Value("${excel.to.database.job.source.file.source.path}")
    private String PROPERTY_EXCEL_SOURCE_FILE_PATH;


    //@Scheduled(cron = "${excel.to.database.job.cron}")
    public void runJob() {
        //String fileSource = PROPERTY_EXCEL_SOURCE_FILE_PATH.concat(PROPERTY_EXCEL_SOURCE_FILE_NAME)+".xlsx";
        Path path = Paths.get(PROPERTY_EXCEL_SOURCE_FILE_PATH);
        WatchKey key;
        WatchService watchService = null;
        try {
            watchService = FileSystems.getDefault().newWatchService();
            path.register(watchService, StandardWatchEventKinds.ENTRY_CREATE);

            while ((key = watchService.take()) != null) {
                for (WatchEvent<?> event : key.pollEvents()) {

                    System.out.println(
                            "Event kind:" + event.kind()
                                    + ". File affected: " + event.context() + ".");
                    if(event.kind().equals("ENTRY_CREATE")) {
                        impoerJob();
                    }
                }
                key.reset();
            }
        } catch (IOException | InterruptedException e) {
            e.printStackTrace();
        }
    }

    private void impoerJob() {

        // make unique JobParameters so now instance of job can be started
        Map<String, JobParameter> confMap = new HashMap<String, JobParameter>();
        confMap.put("time", new JobParameter(System.currentTimeMillis()));
        JobParameters jobParameters = new JobParameters(confMap);
        try {
            JobExecution ex = jobLauncher.run(utilizationBatchConfiguration.importExcelJob(), jobParameters);
            log.info(String.format("Execution status-----> %s, Execution Start Time ------> %s, Execution End Time %s", ex.getStatus(), ex.getStartTime(), ex.getEndTime()));
        } catch (JobExecutionAlreadyRunningException | JobRestartException | JobInstanceAlreadyCompleteException
                | JobParametersInvalidException e) {
            //
            e.printStackTrace();
        }
    }
}

如果我使用调度程序而不是 WatchService，我的批处理作业会成功运行。 这里的问题是，我不知道文件何时会被复制到共享位置。所以必须使用 WatchService 而不是 Scheduler。 我的导入工作是 -

@Component
public class UtilizationBatchConfiguration {
---- removed for brevity

@Bean
    public Job importExcelJob() {
        return jobBuilderFactory.get("importExcelJob")
            .incrementer(new RunIdIncrementer())
            .preventRestart()
            .listener(listener)
            .flow(step1())
            .end()
            .build();
    }
}

这几乎类似于 Spring.io 中的 Spring Batch 服务指南

现在如果我尝试运行代码，我会得到以下响应 -

在 10.506 秒内启动应用程序（JVM 运行时间为 11.676） 2018-07-23 16:55:53.646 信息 18200 --- [线程 2] s.c.a.AnnotationConfigApplicationContext ：关闭 org.springframework.context.annotation.AnnotationConfigApplicationContext@7b9a4292： 启动日期 [2018 年 7 月 23 日星期一 16:55:44 IST]；上下文层次的根 2018-07-23 16:55:53.649 信息 18200 --- [线程 2] o.s.j.e.a.AnnotationMBeanExporter ：取消注册 JMX 公开 bean 关闭 2018-07-23 16:55:53.650 INFO 18200 --- [
线程 2] o.s.j.e.a.AnnotationMBeanExporter ：取消注册 JMX 暴露的 bean 2018-07-23 16:55:53.654 INFO 18200 --- [
线程 2] j.LocalContainerEntityManagerFactoryBean：关闭 JPA 持久性单元“默认”的 EntityManagerFactory 2018-07-23 16:55:53.655 信息 18200 --- [线程 2] com.zaxxer.hikari.HikariDataSource：HikariPool-1 - 关闭 发起... 2018-07-23 16:55:53.683 INFO 18200 --- [线程 2] com.zaxxer.hikari.HikariDataSource：HikariPool-1 - 关闭 完成。

另外，我如何确保每次都在读取新文件，而不是再次读取旧文件？

Answer 1

我不能代表WatchService，但通常JVM 需要至少一个非守护线程运行以防止它关闭。看起来你的应用程序没有这个（所以它正在关闭）。在这个用例中使用 Spring Integration 是很常见的，我们在我们的文档中有这个：https://docs.spring.io/spring-batch/trunk/reference/html/springBatchIntegration.html#launching-batch-jobs-through-messages

Answer 2

您可以这样使用它，唯一的问题是，您的 Spring-Context 在启动后立即关闭，因为“无事可做”。看看this解决方案如何防止你的Spring-Context关闭。

Answer 3

我不完全确定您为什么需要批处理作业功能，因为观察者服务似乎就足够了。

如果您对该位置具有写入权限，您可以标记已处理的文件（可能添加前缀）并更改您的程序以忽略具有该前缀的文件名，或者您可以创建一个已处理的目录并将已处理的文件移动到该目录处理后的目录，或者您可以通过跟踪上次成功的文件进程运行并查找比该时间戳更新的文件来使用时间戳。