Spring Batch项目读取器只执行一次

Question

试图实现Spring批处理，但面临一个奇怪的问题，我们的ItemReader类只执行了一次。

以下是细节。

1. 如果我们在DB中有1000行。
2. 我们的项目阅读器从DB中获取1000行，并将列表传递给ItemWriter
3. ItemWriter成功删除所有项目。
4. 现在ItemReader再次尝试从数据库中获取数据，但没有找到，因此返回NULL，因此执行停止。
5. 但是我们已经将批处理配置为使用Quartz调度器执行，这是每分钟执行一次。
6. 现在，如果我们通过转储导入在DB中插入让我们说1000行，批处理作业应该在下一次执行中选择这个数据，但是它甚至没有执行，尽管JobLauncher正在执行。

配置：-

1.我们有ItemReader和ItemWriter，提交间隔等于1。

<batch:job id="csrfTokenBatchJob">
    <batch:step id="step1">
      <tasklet>
        <chunk reader="csrfTokenReader" writer="csrfTokenWriter" commit-interval="1"></chunk>
      </tasklet>
    </batch:step>
  </batch:job>

2.作业被安排在每一分钟被触发。

<bean class="org.springframework.scheduling.quartz.SchedulerFactoryBean">
    <property name="triggers">
      <bean id="cronTrigger" class="org.springframework.scheduling.quartz.CronTriggerBean">
        <property name="jobDetail" ref="jobDetail" />
        <property name="cronExpression" value="0 0/1 * * * ?" />
      </bean>
    </property>
  </bean>

3.职务配置

<bean id="jobDetail" class="org.springframework.scheduling.quartz.JobDetailBean">
    <property name="jobClass" value="com.tavant.oauth.batch.job.CSRFTokenJobLauncher" />
    <property name="jobDataAsMap">
        <map>
            <entry key="jobName" value="csrfTokenCleanUpBatchJob" />
            <entry key="jobLocator" value-ref="jobRegistry" />
            <entry key="jobLauncher" value-ref="jobLauncher" />
        </map>
    </property>
</bean>

第一次执行成功，但后来没有执行，但我可以在日志中看到JobLauncher正在执行。

@Component("csrfTokenReader")
@Scope(value="step")
public class CSRFTokenReader implements ItemReader<List<CSRFToken>> {

    private static final Logger logger = LoggerFactory.getLogger(CSRFTokenReader.class);

    @Autowired
    private CleanService cleanService;

    @Override
    public List<CSRFToken> read() {
        List<CSRFToken> csrfTokenList = null;
        try{

            int keepUpto = Integer.valueOf(PropertiesContext.getInstance().getProperties().getProperty("token.keep", "1"));

            Calendar calTime = Calendar.getInstance();
            calTime.add(Calendar.HOUR, -keepUpto);
            Date toKeep = calTime.getTime();

            csrfTokenList = cleanService.getCSRFTokenByTime(toKeep);
        }
        catch(Throwable th){
            logger.error("Exception in running job At " + new Date() + th);
        }
        if(CollectionUtils.isEmpty(csrfTokenList)){
            return null;
        }
        return csrfTokenList;
    }
}

编辑：--

public class CSRFTokenJobLauncher extends QuartzJobBean {
    static final String JOB_NAME = "jobName";
    private JobLocator jobLocator;
    private JobLauncher jobLauncher;
    public void setJobLocator(JobLocator jobLocator) {
        this.jobLocator = jobLocator;
    }
    public void setJobLauncher(JobLauncher jobLauncher) {
        this.jobLauncher = jobLauncher;
    }
    @Override
    protected void executeInternal(JobExecutionContext context) {
        Map<String, Object> jobDataMap = context.getMergedJobDataMap();
        String jobName = (String) jobDataMap.get(JOB_NAME);
        log.info("Quartz trigger firing with Spring Batch jobName="+jobName);
        JobParameters jobParameters = getJobParametersFromJobMap(jobDataMap);
        try {
            jobLauncher.run(jobLocator.getJob(jobName), jobParameters);
        }
        catch (JobExecutionException e) {
            log.error("Could not execute job.", e);
        }
    }
    private JobParameters getJobParametersFromJobMap(Map<String, Object> jobDataMap) {
        JobParametersBuilder builder = new JobParametersBuilder();
        for (Entry<String, Object> entry : jobDataMap.entrySet()) {
            String key = entry.getKey();
            Object value = entry.getValue();
            if (value instanceof String && !key.equals(JOB_NAME)) {
                builder.addString(key, (String) value);
            }
            else if (value instanceof Float || value instanceof Double) {
                builder.addDouble(key, ((Number) value).doubleValue());
            }
            else if (value instanceof Integer || value instanceof Long) {
                builder.addLong(key, ((Number)value).longValue());
            }
            else if (value instanceof Date) {
                builder.addDate(key, (Date) value);
            }
        }
        return builder.toJobParameters();
    }
}

Answer 1

Spring batch在数据库中记录每个作业的执行情况。这就是为什么spring batch需要区分每个作业运行。它检查作业是否已在同一天执行，并且不会再次启动，除非任何作业参数与上一次运行不同，或者如果启用了完整设置，则允许启动。

选项1：-如上所述，我们可以使用允许-启动-如果-完成="true"

选项2：-总是传递一个作业参数，它是当前日期时间戳。这样作业参数值总是唯一的。

    JobExecution jobExecution = jobLauncher.run(reportJob, new JobParametersBuilder()
                    .addDate("now", new Date()).build());

选项3：-使用增量管理器，例如RunIdCrementer，这样我们就不需要确保每次都传递唯一的作业参数。

    @Bean
    public Job job1(JobBuilderFactory jobs, Step s1) {
        return jobs.get("job1")
                .incrementer(new RunIdIncrementer())
                .flow(s1)
                .end()
                .build();
    }

Answer 2

经过几个小时的时间浪费，这个问题似乎已经解决了，我已经在tasklet中配置了允许-启动-如果-完成="true"。现在批处理项目阅读器正在按照计划执行。

<batch:job id="csrfTokenBatchJob">
    <batch:step id="step1">
      <batch:tasklet allow-start-if-complete="true">
        <batch:chunk reader="csrfTokenReader" writer="csrfTokenWriter" commit-interval="1"></batch:chunk>
      </batch:tasklet>
    </batch:step>
  </batch:job>