Spring Batch在步骤之间共享大量数据

Question

我需要在作业步骤之间共享相对大量的数据以实现 Spring 批处理。我知道StepExecutionContext 和JobExecutionContext 作为这种机制。但是，我读到了，因为它们的大小必须受到限制（少于 2500 个字符）。这对我的需要来说太小了。在我的新手一步Spring Batch 实现中，我的单步工作如下：

@Configuration
@EnableBatchProcessing
public class BatchConfig {

    @Autowired
    public JobBuilderFactory jobBuilderFactory;

    @Autowired
    public StepBuilderFactory stepBuilderFactory;


    private static final String GET_DATA =
    "    SELECT " +
    "stuffA, " +
    "stuffB, " +
    "FROM STUFF_TABLE " +
    "ORDER BY stuffA ASC";

    @Bean
    public ItemReader<StuffDto> databaseCursorItemReader(DataSource dataSource) {
        return new JdbcCursorItemReaderBuilder<StuffDto>()
                .name("cursorItemReader")
            .dataSource(dataSource)
            .sql(GET_DATA)
            .rowMapper(new BeanPropertyRowMapper<>(StuffDto.class))
                .build();
    }

    @Bean
    ItemProcessor<StuffDto, StuffDto> databaseXmlItemProcessor() {
        return new QueryLoggingProcessor();
    }

    @Bean
    public ItemWriter<StuffDto> databaseCursorItemWriter() {
        return new LoggingItemWriter();
    }

    @Bean
    public Step databaseCursorStep(@Qualifier("databaseCursorItemReader") ItemReader<StuffDto> reader,
    @Qualifier("databaseCursorItemWriter") ItemWriter<StuffDto> writer,
    StepBuilderFactory stepBuilderFactory) {
        return stepBuilderFactory.get("databaseCursorStep")
            .<StuffDto, StuffDto>chunk(1)
        .reader(reader)
            .writer(writer)
            .build();
    }

    @Bean
    public Job databaseCursorJob(@Qualifier("databaseCursorStep") Step exampleJobStep,
    JobBuilderFactory jobBuilderFactory) {
        return jobBuilderFactory.get("databaseCursorJob")
            .incrementer(new RunIdIncrementer())
            .flow(exampleJobStep)
            .end()
            .build();
    }
}

从某种意义上说，这很好用，我可以成功地从数据库中读取数据并在写入器步骤中写入这样的 loggingitemwriter：

public class LoggingItemWriter implements ItemWriter<StuffDto> {

    private static final Logger LOGGER = LoggerFactory.getLogger(LoggingItemWriter.class);

    @Override
    public void write(List<? extends StuffDto> list) throws Exception {
        LOGGER.info("Writing stuff: {}", list);
    }
}

但是，我需要能够将 StuffDto（或等效项）和它的数据提供给第二步，该步骤将对它执行一些处理，而不仅仅是记录它。

如果您认为步骤和工作环境过于有限，我将不胜感激。谢谢。

Answer 1

如果您不想在数据库或文件系统中写入数据，实现相同的一种方法如下：

1. 在您的 config 类中创建您自己的作业上下文 bean，并使用 @JobScope 对其进行注释。
2. 为您的阅读器、处理器和编写器类实现org.springframework.batch.core.step.tasklet 接口。如果您想更好地控制步骤，您也可以使用它实现org.springframework.batch.core.StepExecutionListener。
3. 使用@Autowire 获取您自己的context 对象，并使用它的setter-getter 方法来存储和检索数据。

示例代码：

Config.java

@Autowired
private Processor processor;

@Autowired
private Reader reader;

@Autowired
private Writer writer;

@Autowired
private JobBuilderFactory jobBuilderFactory;
     
@Autowired
private StepBuilderFactory stepBuilderFactory;

@Bean
@JobScope
public JobContext getJobContexts() {
    JobContext newJobContext = new JobContext();
    return newJobContext;
}

@Bean
public Step reader() {
      return stepBuilderFactory.get("reader")
        .tasklet(reader)
        .build();
     }

@Bean
public Step processor() {
      return stepBuilderFactory.get("processor")
        .tasklet(processor)
        .build();
     }

@Bean
public Step writer() {
      return stepBuilderFactory.get("writer")
        .tasklet(writer)
        .build();
     }


public Job testJob() {
      
      return jobBuilderFactory.get("testJob")
        .start(reader())
        .next(processor())
        .next(writer())
        .build();
     }

//Below will start the job
@Scheduled(fixedRate = 1000)
    public void starJob(){
        
        Map<String, JobParameter> confMap = new HashMap<>();
        confMap.put("time", new JobParameter(new Date()));
        JobParameters jobParameters = new JobParameters(confMap);
        monitorJobLauncher.run(testJob(), jobParameters);
        
}

JobContext.java

private List<StuffDto> dataToProcess = new ArrayList<>();
private List<StuffDto> dataToWrite = new ArrayList<>();

//getter

SampleReader.java

@Component
public class SampleReader  implements Tasklet,StepExecutionListener{
        @Autowired
        private JobContext context;
       
      @Override
      public void beforeStep(StepExecution stepExecution) {
    //logic that you need to perform before the execution of this step.
      }
    
    @Override
      public void afterStep(StepExecution stepExecution) {
    //logic that you need to perform after the execution of this step.
      }
    
    @Override
    public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext){
     // Whatever code is here will get executed for reader.
     //  Fetch StuffDto object from database and add it to jobContext 
   //dataToProcess list.
    return RepeatStatus.FINISHED;
    }
}

SampleProcessor.java

   @Component
   public class SampleProcessor  implements Tasklet{
    
       @Autowired
       private JobContext context;
    
        @Override
        public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext){
        
         // Whatever code is here will get executed for processor.
        // context.getDataToProcessList();
       // apply business logic and set the data to write.

return RepeatStatus.FINISHED;
    }

writer 类的方法相同。

注意：这里请注意，这里需要自己编写数据库相关的样板代码。但是通过这种方式，您可以更好地控制您的逻辑，而无需担心上下文大小限制。所有数据都将在内存中，因此一旦操作完成，这些数据将被垃圾收集。我希望你能明白我愿意传达的意思。

要了解更多关于 Tasklet 与 Chunk 的信息，请阅读 this。