Java 类org.springframework.batch.core.step.tasklet.TaskletStep 实例源码

@Override
public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {

    registerJobExecutionEventListener(bean);

    if (bean instanceof AbstractStep) {
        registerStepExecutionEventListener(bean);
        if (bean instanceof TaskletStep) {
            TaskletStep taskletStep = (TaskletStep) bean;
            Tasklet tasklet = taskletStep.getTasklet();
            registerChunkEventsListener(bean);

            if (tasklet instanceof ChunkOrientedTasklet) {
                Field chunkProviderField = ReflectionUtils.findField(ChunkOrientedTasklet.class, "chunkProvider");
                ReflectionUtils.makeAccessible(chunkProviderField);
                SimpleChunkProvider chunkProvider = (SimpleChunkProvider) ReflectionUtils.getField(chunkProviderField, tasklet);
                Field chunkProcessorField = ReflectionUtils.findField(ChunkOrientedTasklet.class, "chunkProcessor");
                ReflectionUtils.makeAccessible(chunkProcessorField);
                SimpleChunkProcessor chunkProcessor = (SimpleChunkProcessor) ReflectionUtils.getField(chunkProcessorField, tasklet);
                registerItemReadEvents(chunkProvider);
                registerSkipEvents(chunkProvider);
                registerItemProcessEvents(chunkProcessor);
                registerItemWriteEvents(chunkProcessor);
                registerSkipEvents(chunkProcessor);
            }
        }
    }

    return bean;
}

private void registerChunkEventsListener(Object bean) {
    if(this.applicationContext.containsBean(BatchEventAutoConfiguration.CHUNK_EVENTS_LISTENER))
    {
        ((TaskletStep)bean).registerChunkListener((ChunkListener)
                this.applicationContext.getBean(BatchEventAutoConfiguration.CHUNK_EVENTS_LISTENER));
    }
}

@Test
public void testPostProcessor() {
    TaskBatchEventListenerBeanPostProcessor postProcessor =
            context.getBean(TaskBatchEventListenerBeanPostProcessor.class);
    assertNotNull(postProcessor);
    TaskletStep updatedTaskletStep = (TaskletStep)
            postProcessor.postProcessBeforeInitialization(taskletStep, "FOO");
    assertEquals(taskletStep, updatedTaskletStep);
}

protected TaskletStep createTaskletStep(String name, Tasklet tasklet) {
    TaskletStep ts = new TaskletStep(name);
    ts.setJobRepository(jobRepository);
    ts.setTransactionManager(transactionManager);
    ts.setTasklet(tasklet);

    return ts;
}

@Override
protected void executeInternal(JobExecutionContext context) throws JobExecutionException {

    TaskletStep ts = createTaskletStep("snapshotRun", snapshotBatchJobTasklet);

    SimpleJob job = createSimpleJob("snapshotBatchJob", ts);

    // by default, jobs are assumed to have been already run if the exact same
    // parameters are seen. Pass in the time to always be unique.
    Map<String, JobParameter> map = new HashMap<>();
    map.put("executeTime", new JobParameter(System.currentTimeMillis()));

    runJob(job, context, map);
}

@Override
protected void executeInternal(JobExecutionContext context) throws JobExecutionException {

    TaskletStep ts = createTaskletStep("nightlyRun", nightlyReportBatchJobTasklet);

    SimpleJob job = createSimpleJob("nightlyReportBatchJob", ts);

    // by default, jobs are assumed to have been already run if the exact same
    // parameters are seen. Pass in the time to always be unique.
    Map<String, JobParameter> map = new HashMap<>();
    map.put("executeTime", new JobParameter(System.currentTimeMillis()));

    runJob(job, context, map);
}

@Override
protected void executeInternal(JobExecutionContext context) throws JobExecutionException {

    TaskletStep ts = createTaskletStep("HourlyRun", HourlyReportBatchJobTasklet);

    SimpleJob job = createSimpleJob("HourlyReportBatchJob", ts);

    // by default, jobs are assumed to have been already run if the exact same
    // parameters are seen. Pass in the time to always be unique.
    Map<String, JobParameter> map = new HashMap<>();
    map.put("executeTime", new JobParameter(System.currentTimeMillis()));

    runJob(job, context, map);
}

@Override
protected void executeInternal(JobExecutionContext context) throws JobExecutionException {

    TaskletStep ts = createTaskletStep("deleteSnapshotWindow", deleteSnapshotBatchJobTasklet);

    SimpleJob job = createSimpleJob("deleteSnapshotWindowBatchJob", ts);

    // by default, jobs are assumed to have been already run if the exact same
    // parameters are seen. Pass in the time to always be unique.
    Map<String, JobParameter> map = new HashMap<>();
    map.put("executeTime", new JobParameter(System.currentTimeMillis()));
    map.put("snapshotWindowId", new JobParameter(context.getJobDetail().getJobDataMap().getLong("snapshotWindowId")));

    runJob(job, context, map);
}

@Bean
@Autowired
protected Step ccJudgmentImportProcessStep() {
    TaskletStep step = steps.get("ccJudgmentImportProcessStep")
            .<RawSourceCcJudgment, JudgmentWithCorrectionList<CommonCourtJudgment>> chunk(20).faultTolerant()
           .skipPolicy(ccjImportProcessSkipPolicy).listener(ccjImportProcessSkipListener)
           .reader(ccjImportProcessReader)
           .processor(ccjImportProcessProcessor)
           .writer(ccjImportProcessWriter())
           .transactionAttribute(new DefaultTransactionAttribute())
           .build();
       step.setTransactionAttribute(new DefaultTransactionAttribute());
       return step;
}