将数据传递给未来的步骤 - Spring Batch

Question

正如我们在 Spring Batch 官方文档中看到的那样，“将数据传递到未来的步骤 - Spring Batch”是可能的，但我们大多数人一直在努力解决它，因为在官方文档中他们提到了两种可能性。一个台阶级别，一个工作级别。问题是如何检索步骤级别的数据？

我的解决方案与官方文档中的相关解决方案相同，但它不起作用。所以我决定做以下事情：

1-为促销监听器创建bean：

<beans:bean id="promotionListener"
                class="org.springframework.batch.core.listener.ExecutionContextPromotionListener">
        <beans:property name="keys" value="someValues"/>
</beans:bean>

2-在您的步骤中设置监听器（您希望保存数据的步骤）

        <listeners>
            <listener ref="promotionListener"/>
        </listeners>

3- 在写入器中的同一步骤（将保存数据的步骤）中，保存数据。

private StepExecution stepExecution;

    @BeforeStep
    public void saveStepExecution(StepExecution stepExecution) {
        this.stepExecution = stepExecution;

        ExecutionContext executionContext = stepExecution.getExecutionContext();
        Map<String, String> fieldsMap= new HashMap<>();
        executionContext.put("keys", someValues);
    }

@Override
public void write(List<? extends Map<String, String>> items) throws Exception {
    LOGGER.info(items.toString());

    Map<String, String> fieldsMap= new ConcurrentHashMap<>();
    items.forEach(item -> item.forEach(fieldsMap::put));

    ExecutionContext stepContext = this.stepExecution.getExecutionContext();
    stepContext.put("keys", fieldsMap);
}

您可以看到，在我的例子中，我将数据保存在 Map (ConcurrentHashMap) 中。

4-重要提示：在下一步中，您想要检索我们在上一步中保存的数据。按照这个顺序，我们必须做：

• 声明将保存我们将检索的值的对象：

  私有地图字段Map；

关注JobExecution

@BeforeStep
    public void retrieveInterStepData(StepExecution stepExecution) {
        JobExecution jobExecution = stepExecution.getJobExecution();
        Collection<StepExecution> stepExecutions = jobExecution.getStepExecutions();
        for (StepExecution steps : stepExecutions) {
            ExecutionContext executionContext = steps.getExecutionContext();
            if (executionContext.containsKey("keys")) {
                this.nationalityMap = (Map<String, String>) executionContext.get("keys");
            }
        }
    }

就是这样！你可能会奇怪为什么我没有按照官方文档的写法来操作呢？原因是我在同一个工作中与 Steps 一起工作。他们共享相同的工作执行。现在看一下我的调试模式的图片。

如果还有其他方法请建议。

注意：请不要只是复制并粘贴官方文档中的代码，而是提供您自己的答案或实现。

与此问题相关的 Spring Batch 文档的链接如下， 在此处输入链接描述

Answer 1

您将要从步骤执行上下文提升到作业执行上下文的键与数据本身混淆了。这种混乱来自两个地方：

• <beans:property name="keys" value="someValues"/>：someValues应该someKeys
• 调用executionContext.put("keys", someValues);不@BeforeStep正确

让我说得更清楚一些。想象一下你的工作有两个步骤：

• 步骤 1：读取/写入一些项目并将项目计数写入步骤执行上下文中
• 步骤 2：需要访问项目计数并将其打印到控制台。

在这种情况下，您可以使用提升侦听器将键“count”从步骤 1 的步骤执行上下文提升到作业执行上下文，以便步骤 2 可以访问它。这是一个例子：

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.beans.factory.annotation.Autowired;
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 {

    @Autowired
    private JobBuilderFactory jobs;

    @Autowired
    private StepBuilderFactory steps;

    @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) {
                for (Integer item : items) {
                    System.out.println("item = " + item);
                }
                ExecutionContext stepContext = this.stepExecution.getExecutionContext();
                int count = stepContext.containsKey("count") ? stepContext.getInt("count") : 0;
                stepContext.put("count", count + items.size());
            }

            @BeforeStep
            public void saveStepExecution(StepExecution stepExecution) {
                this.stepExecution = stepExecution;
            }
        };
    }

    @Bean
    public Step step1() {
        return steps.get("step1")
                .<Integer, Integer>chunk(2)
                .reader(itemReader())
                .writer(itemWriter())
                .listener(promotionListener())
                .build();
    }

    @Bean
    public Step step2() {
        return steps.get("step2")
                .tasklet((contribution, chunkContext) -> {
                    // retrieve the key from the job execution context
                    Integer count = (Integer) chunkContext.getStepContext().getJobExecutionContext().get("count");
                    System.out.println("In step 2: step 1 wrote " + count + " items");
                    return RepeatStatus.FINISHED;
                })
                .build();
    }

    @Bean
    public ExecutionContextPromotionListener promotionListener() {
        ExecutionContextPromotionListener listener = new ExecutionContextPromotionListener();
        listener.setKeys(new String[] {"count"});
        return listener;
    }

    @Bean
    public Job job() {
        return jobs.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
In step 2: step 1 wrote 4 items

希望这可以帮助。