如何从 Spring Boot Web 应用程序中 Spring 批处理作业的 ItemReader 类访问作业参数

Question

我正在从我的控制器启动一个春季批处理作业。我需要通过基于 java 的配置从配置的自定义ItemReader 访问作业参数fileName。但我总是从我的 ItemReader 获取文件名为 null。请帮忙。 我之所以这样启动，是因为我必须根据用户请求更改 ItemReader 文件。

我的控制器：

@RestController
@RequestMapping(value = "/")
public class SimpleController {


@Autowired
Step step;

@Autowired
private JobBuilderFactory jobs;

@Autowired
private StepBuilderFactory steps;

@Autowired
private JobRepository jobRepository;

@Autowired
JobLauncher jobLauncher;

@RequestMapping(method = RequestMethod.GET)
public void StartJob() throws JobParametersInvalidException, JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException {
    JobParameter param = new JobParameter("ValueBeingSetFromRequestHeader");
    HashMap map = new HashMap<String, JobParameter>();
    map.put("fileName",param);
    SimpleJob job = new SimpleJob("test");
    job.addStep(step);
    System.out.println(jobRepository);
    job.setJobRepository(jobRepository);

    System.out.println( jobRepository.getLastJobExecution(job.getName(), new JobParameters(map)));

    jobLauncher.run(job, new JobParameters(map));

}

}

而我的 Job 步骤配置类是

public class ProducerBatchJobConfigurer {


private String fileName;

public String getFileName() {
    return fileName;
}
@Value("#{jobParameters['fileName']}")
public void setFileName(String fileName) {
    this.fileName = fileName;
}

@Bean
public ItemReader reader() {
    System.out.println(fileName);
    if (fileName == null)
        return new DummyReader();
    FlatFileItemReader faltFileReader =  new FlatFileItemReader();

    faltFileReader.setResource(new FileSystemResource(new File(fileName)));
    faltFileReader.setResource(new FileSystemResource(new File(fileName)));
    DefaultLineMapper<KafkaMessage> lineMapper = new DefaultLineMapper<KafkaMessage>();
    lineMapper.setLineTokenizer(new DelimitedLineTokenizer());
    lineMapper.setFieldSetMapper(new CSVFieldMapper());
    faltFileReader.setLineMapper(lineMapper);
    return faltFileReader;
}

@Bean
public ItemProcessor processor() {
   return new CustomItemProcessor();

}

@Bean
public ItemWriter writer() {
    return new KafkaWriter();
}



@Bean
public Step step1(StepBuilderFactory stepBuilderFactory,
                  ItemReader reader, ItemWriter writer,
                  ItemProcessor processor) {
    System.out.println("ddddddddddddddd");
    return stepBuilderFactory.get("step1")
            .chunk(10).reader(reader)
            .processor(processor).writer(writer).build();
}
@Bean
public StepScope stepScope() {
    final StepScope stepScope = new StepScope();
    stepScope.setAutoProxy(true);
    return stepScope;
}
@Bean
public JobScope jobScope(){
    return new JobScope();
}
}

Answer 1

@Bean
public ItemReader reader() {
 ...
}

应该改为

@Bean
public ItemStreamReader reader() {
 ...
}

甚至更具体

@Bean
public FlatFileItemReader reader() {
 ...
}

步骤定义也应更改为

@Bean
public Step step1(StepBuilderFactory stepBuilderFactory,
                  ItemStreamReader reader, ItemWriter writer,
                  ItemProcessor processor) {

}

为什么需要这样做？

如果 itemReader 实现了ItemStream 接口（或ItemStreamReader），Spring 会这样对待它。

见https://docs.spring.io/spring-batch/reference/html/readersAndWriters.html#delegatePatternAndRegistering

直接连接到 Step 的读取器、写入器或处理器 如果它实现 ItemStream 或 StepListener 接口。

这意味着 spring 将调用 open、update 和 close 方法

Spring Batch: org.springframework.batch.item.ReaderNotOpenException: Reader must be open before it can be read的类似问题

Answer 2

使用@Value("#{jobParameters[fileName]}") 而不是@Value("#{jobParameters['fileName']}")