spring - 在 Spring Batch 中的步骤之间传递值时出现空指针异常
问题描述
我正在学习 Spring Batch,我正在做一个示例程序,我需要在其中将值从一个步骤传递到另一个步骤。
场景:我有一个人员表,我从中提取人员详细信息,将几列保存到 DTO(在步骤 1 的 ItemWriter 中)并将值从 DTO 传递到 where 子句上的另一个表从中提取相关值(在第 2 步的 ItemReader 中)。最后,我将生成一个包含所有这些值的 CSV。
这是我的代码:
@Bean
public Job job() throws UnexpectedInputException, ParseException, NonTransientResourceException, Exception {
return jobBuilderFactory.get("readDBJob").incrementer(new RunIdIncrementer()).start(step1()).next(step2())
.build();
}
@Bean
public Step step1() {
return stepBuilderFactory.get("step1").<Person, Person>chunk(500000).reader(itemReader())
.writer(itemWriter()).listener(promotionListener()).build();
}
@Bean
public Step step2() throws UnexpectedInputException, ParseException, NonTransientResourceException, Exception {
return stepBuilderFactory.get("step2").<Person, Result>chunk(100)
.reader(readingObjectItemReader.cursorReader()).writer(itemWriterForStep2()).build();
}
第 1 步的 ItemWriter:
@Bean
public ItemWriter<Person> itemWriter() {
return new ItemWriter<Person>() {
private StepExecution stepExecution;
List<personDTO> responseList = null;
@Override
public void write(List<? extends Person> items) throws Exception {
for (Person item : items) {
personDTO responseObject = new personDTO();
BeanUtils.copyProperties(item, responseObject);
if(responseObject != null && responseObject.getPersonId() != null) {
if(stepExecution.getExecutionContext().containsKey("personDtoObject")) {
responseList = (List<personDTO>) this.stepExecution.getExecutionContext().get("personDtoObject");
}
responseList.add(responseObject);
this.stepExecution.getExecutionContext().put("personDtoObject", responseList);
}
}
}
@BeforeStep
public void saveStepExecution(StepExecution stepExecution) {
this.stepExecution = stepExecution;
this.stepExecution.getExecutionContext().put("personDtoObject", new ArrayList<>());
}
}
作业执行上下文:
@Bean
public Object promotionListener() {
ExecutionContextPromotionListener listener = new ExecutionContextPromotionListener();
listener.setKeys(new String[] {"personDtoObject"});
listener.setStrict(true);
return listener;
}
这是我尝试在步骤 2 ItemReader 中访问值的方式
公共类 ReadingObjectItemReader 实现 ItemReader {
@Autowired
DataSource dataSource;
private List<personDTO> personDtoList;
String value;
@Override
public personDetails read()
throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {
return null;
}
@Bean
public JdbcCursorItemReader<personDetails> cursorReader() {
System.out.println("Values from the step 1 " + personDtoList);
....
}
@BeforeStep
public void retrieveSharedData(StepExecution stepExecution) {
JobExecution jobExecution = stepExecution.getJobExecution();
ExecutionContext jobContext = jobExecution.getExecutionContext();
personDtoList= (List<personDTO>) jobContext.get("personDtoObject");
}
}
当我尝试在第 2 步中访问 personDtoList 的值时,我得到了 null。我在步骤 1 完成之前验证了 StepContext 中的值,一切看起来都很好,但是当尝试在步骤 2 中访问它们时,我得到了 null。
我查看了大多数在线可用资源,但我无法弄清楚我哪里出错了。任何帮助表示赞赏。
我在这里先向您的帮助表示感谢。
解决方案
在步骤 1 的项目编写器中,您正在执行以下操作:
ExecutionContext stepContext = this.stepExecution.getExecutionContext();
stepContext.put("personDtoObject", responseList);
这意味着您将覆盖每个块的先前列表。您需要做的是从执行上下文中获取列表并在覆盖键之前在其中添加项目。您还需要在步骤边界(也就是第一个块和最后一个块)处添加一些完整性检查,以确保列表已初始化并且它不是null
在将其放入执行上下文之前(尤其是最后一个块)。
编辑:添加促销侦听器工作所需的代码更改
您还需要将promotionListener()
方法的返回类型从更改Object
为ExecutionContextPromotionListener
:
@Bean
public ExecutionContextPromotionListener promotionListener() {
ExecutionContextPromotionListener listener = new ExecutionContextPromotionListener();
listener.setKeys(new String[] {"personDtoObject"});
listener.setStrict(true);
return listener;
}
否则,此 bean 未正确注册为侦听器。这是一个完整的例子:
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.annotation.BeforeStep;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.listener.ExecutionContextPromotionListener;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.support.ListItemReader;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
@EnableBatchProcessing
public class MyJob {
private JobBuilderFactory jobBuilderFactory;
private StepBuilderFactory stepBuilderFactory;
public MyJob(JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory) {
this.jobBuilderFactory = jobBuilderFactory;
this.stepBuilderFactory = stepBuilderFactory;
}
@Bean
public ItemReader<Integer> itemReader() {
return new ListItemReader<>(Arrays.asList(1, 2, 3, 4));
}
@Bean
public ItemWriter<Integer> itemWriter() {
return new ItemWriter<Integer>() {
private StepExecution stepExecution;
@Override
public void write(List<? extends Integer> items) {
List<Integer> itemsList = (List<Integer>) stepExecution.getExecutionContext().get("items");
for (Integer item : items) {
System.out.println("item = " + item);
itemsList.add(item);
}
}
@BeforeStep
public void saveStepExecution(StepExecution stepExecution) {
this.stepExecution = stepExecution;
this.stepExecution.getExecutionContext().put("items", new ArrayList<>());
}
};
}
@Bean
public Step step1() {
return stepBuilderFactory.get("step1")
.<Integer, Integer>chunk(2)
.reader(itemReader())
.writer(itemWriter())
.listener(promotionListener())
.build();
}
@Bean
public Step step2() {
return stepBuilderFactory.get("step2")
.tasklet((contribution, chunkContext) -> {
ExecutionContext executionContext = contribution.getStepExecution().getJobExecution().getExecutionContext();
List<Integer> items = (List<Integer>) executionContext.get("items");
System.out.println("Items read in step1:");
for (Integer item : items) {
System.out.println("item = " + item);
}
return RepeatStatus.FINISHED;
})
.build();
}
@Bean
public ExecutionContextPromotionListener promotionListener() {
ExecutionContextPromotionListener listener = new ExecutionContextPromotionListener();
listener.setKeys(new String[]{"items"});
listener.setStrict(true);
return listener;
}
@Bean
public Job job() {
return jobBuilderFactory.get("job")
.start(step1())
.next(step2())
.build();
}
public static void main(String[] args) throws Exception {
ApplicationContext context = new AnnotationConfigApplicationContext(MyJob.class);
JobLauncher jobLauncher = context.getBean(JobLauncher.class);
Job job = context.getBean(Job.class);
jobLauncher.run(job, new JobParameters());
}
}
打印:
item = 1
item = 2
item = 3
item = 4
Items read in step1:
item = 1
item = 2
item = 3
item = 4
编辑 2:添加面向块的步骤的示例
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.function.Consumer;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.annotation.BeforeStep;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.listener.ExecutionContextPromotionListener;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.support.ListItemReader;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
@EnableBatchProcessing
public class MyJob {
private JobBuilderFactory jobBuilderFactory;
private StepBuilderFactory stepBuilderFactory;
public MyJob(JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory) {
this.jobBuilderFactory = jobBuilderFactory;
this.stepBuilderFactory = stepBuilderFactory;
}
@Bean
public ItemReader<Integer> itemReader() {
return new ListItemReader<>(Arrays.asList(1, 2, 3, 4));
}
@Bean
public ItemWriter<Integer> itemWriter() {
return new ItemWriter<Integer>() {
private StepExecution stepExecution;
@Override
public void write(List<? extends Integer> items) {
List<Integer> itemsList = (List<Integer>) stepExecution.getExecutionContext().get("items");
for (Integer item : items) {
System.out.println("item = " + item);
itemsList.add(item);
}
}
@BeforeStep
public void saveStepExecution(StepExecution stepExecution) {
this.stepExecution = stepExecution;
this.stepExecution.getExecutionContext().put("items", new ArrayList<>());
}
};
}
@Bean
public Step step1() {
return stepBuilderFactory.get("step1")
.<Integer, Integer>chunk(2)
.reader(itemReader())
.writer(itemWriter())
.listener(promotionListener())
.build();
}
@Bean
public Step step2() {
return stepBuilderFactory.get("step2")
.<Integer, Integer>chunk(2)
.reader(new ReadingObjectItemReader())
.writer(items -> items.forEach((Consumer<Integer>) integer -> System.out.println("integer = " + integer)))
.build();
}
@Bean
public ExecutionContextPromotionListener promotionListener() {
ExecutionContextPromotionListener listener = new ExecutionContextPromotionListener();
listener.setKeys(new String[]{"items"});
listener.setStrict(true);
return listener;
}
@Bean
public Job job() {
return jobBuilderFactory.get("job")
.start(step1())
.next(step2())
.build();
}
public static void main(String[] args) throws Exception {
ApplicationContext context = new AnnotationConfigApplicationContext(MyJob.class);
JobLauncher jobLauncher = context.getBean(JobLauncher.class);
Job job = context.getBean(Job.class);
jobLauncher.run(job, new JobParameters());
}
public static class ReadingObjectItemReader implements ItemReader<Integer> {
int i = 0;
private List<Integer> items;
@Override
public Integer read() {
if (i >= items.size()) {
return null;
} else {
return items.get(i++);
}
}
@BeforeStep
public void retrieveSharedData(StepExecution stepExecution) {
JobExecution jobExecution = stepExecution.getJobExecution();
ExecutionContext jobContext = jobExecution.getExecutionContext();
items = (List<Integer>) jobContext.get("items");
}
}
}
印刷:
item = 1
item = 2
item = 3
item = 4
integer = 1
integer = 2
integer = 3
integer = 4
这意味着该列表已在步骤 2 中从作业执行上下文中正确检索,这是一个面向块的步骤。
推荐阅读
- html - 如何为网站搜索结果添加图像元标记关键字?
- python - 如何将子查询 SQL 转换为 sqlachemy?
- java - 如何识别 HashMap 中的哪个值给出了一个真值,然后用它做些什么?
- gcc - 为什么`mov %eax, %eax; nop` 比`nop` 快?
- mysql - mysql:从2个不同的表中选择所有行
- python - 风险指数超出范围
- php - AWS EC2 PHP:无法设置会话值,但正在生成会话 ID
- matplotlib - Matplotlib 热图失败
- firebase - 在没有 App Server 的情况下发送 FCM 通知
- python - Spark - 将 DataFrame 转换为 RowMatrix 以有效地计算所有对相似度