从数据库获取数据后动态设置chunksize

Question

我需要在spring批处理作业的步骤中动态设置块大小，该步骤存储在数据库中，即需要从数据库中获取块大小并将其设置到bean中。

我的问题是：

从ID='some_id_param_value'的SOME_TABLE_NAME选择CHUNK_SIZE

在这里，ID的值将来自作业参数，该参数是通过与请求一起传递到Rest控制器的请求参数设置的（触发批处理作业时）

我想从数据库中获取这个CHUNK_SIZE，并将其动态设置到作业步骤中。我们的要求是，chunksize根据ID值而变化，ID值的详细信息存储在db表中。例如：

我知道作业中的bean是在配置时设置的，作业参数是在触发作业时在运行时传递的。

编辑：

MahmoudBenHassine提供的示例使用@JobScope并使用@Value（“#{jobParameters['id']}”）访问步骤bean中的作业参数。我尝试使用jobExecutionContext实现类似的方法，如下所示：

>

用@JobScope注释step bean，并使用@Value（“#{jobExecutionContext['chunk']}”）在step bean中访问它。

但我面临以下错误：

创建名为“scopedTarget”的bean时出错。在类路径资源[com/sample/config/SampleBatchConfig.class]中定义的步骤：通过工厂方法实例化Bean失败；嵌套的异常是org。springframework。豆。BeanInstationException:未能实例化[org.springframework.batch.core.Step]：工厂方法“Step”引发异常；嵌套的例外是java。lang.NullPointerException

它无法从jobExecutionContext访问“chunk”键值，因此引发NullPointerException。是否需要以某种方式对其进行升级，以便可以在step bean中访问它？如果是的话，请提供快速样品或指导。

我的控制器类：

@RestController
public class SampleController {

    @Autowired
    JobLauncher sampleJobLauncher;

    @Autowired
    Job sampleJob;
    
    @GetMapping("/launch")
    public BatchStatus launch(@RequestParam(name = "id", required = true) String id){

        Map<String, JobParameter> map = new HashMap<>();
        map.put("id",  new JobParameter(id));
        map.put("timestamp",  new JobParameter(System.currentTimeMillis));

        JobParameters params = new JobParameters(map);
        JobExecution j = sampleJobLauncher.run(sampleJob, params);

        return j.getStatus();
    }
}   

我的批处理配置类（包含作业和步骤bean）：

@Configuration
public class SampleBatchConfig{

    @Autowired
    private JobBuilderFactory myJobBuilderFactory;

    @Autowired
    private StepBuilderFactory myStepBuilderFactory;

    @Autowired
    private MyRepoClass myRepo; // this class contains the jdbc method to fetch chunksize from the db table
    
    @Autowired
    MyReader myReader;
    
    @Autowired
    MyWriter myWriter;
    
    @Bean
    @JobScope
    public Step sampleStep(@Value("#{jobExecutionContext['chunk']}") Integer chunkSize){
        return myStepBuilderFactory.get("sampleStep")
                .<MyClass, MyClass>chunk(chunkSize) //TODO ~instead of hardcoding the chunkSize or getting it from the properties file using @Value, the requirement is to fetch it from the db table using the above mentioned query with id job parameter and set it here
                .reader(myReader.sampleReader())
                .writer(myWriter.sampleWriter())
                .listener(new StepExecutionListener() {
                    @Override
                    public void beforeStep(StepExecution stepExecution) {
                        int chunk = myRepo.findChunkSize(stepExecution.getJobExecution().getExecutionContext().get("id")); // this method call fetches chunksize from the db table using the id job parameter
                        stepExecution.getJobExecution().getExecutionContext().put("chunk", chunk);
                    }

                    @Override
                    public ExitStatus afterStep(StepExecution stepExecution) {
                        return null;
                    }
                })
                .build();
    }

    @Bean
    public Job job(){
        return myJobBuilderFactory.get("sampleJob")
                .incrementer(new RunIdIncrementer())
                .start(sampleStep(null))
                .build();
    }

}

注意：作业可能有多个具有不同chunkSize的步骤，在这种情况下，chunkSize将针对每个步骤单独获取。

编辑2：如下更改我的步骤定义是可行的，但有一个问题。在这里，读者阅读一个包含17项的列表，大小为4。

@Bean
@JobScope
public Step sampleStep(@Value("#{jobParameters['id']}") Integer id){
   int chunkSize = myRepo.findChunkSize(id); // this method call fetches chunksize from the db table using the id job parameter
   return myStepBuilderFactory.get("sampleStep")
                .<MyClass, MyClass>chunk(chunkSize)
                .reader(myReader.sampleReader())
                .writer(myWriter.sampleWriter())  
                .listener(new ChunkListenerSupport() {
                    @Override
                    public void afterChunk(ChunkContext context) {
                        System.out.println("MyJob.afterChunk");
                    }

                    @Override
                    public void beforeChunk(ChunkContext context) {
                        System.out.println("MyJob.beforeChunk");
                    }
                })                      
                .build();
}

我第一次从url触发作业时，它工作正常并打印以下内容：（db表中的块大小设置为4）

2021-05-03 15:06:44.859  INFO 11924 --- [nio-8081-exec-1] o.s.batch.core.job.SimpleStepHandler     : Executing step: [sampleStep]
MyJob.beforeChunk

item = 1

item = 2

item = 3

item = 4

MyJob.afterChunk

MyJob.beforeChunk

item = 5

item = 6

item = 7

item = 8

MyJob.afterChunk

MyJob.beforeChunk

item = 9

item = 10

item = 11

item = 12

MyJob.afterChunk

MyJob.beforeChunk

item = 13

item = 14

item = 15

item = 16

MyJob.afterChunk

MyJob.beforeChunk

item = 17

MyJob.afterChunk

但是如果我再次触发作业，而不重新启动服务器/Spring容器，则打印以下内容：

2021-05-03 15:11:02.427  INFO 11924 --- [nio-8081-exec-4] o.s.batch.core.job.SimpleStepHandler     : Executing step: [sampleStep]

MyJob.beforeChunk

MyJob.afterChunk

简而言之，当服务器重新启动时，它只能正常工作一次。但是在不重新启动服务器的情况下，它不适用于后续的作业执行。

Answer 1

由于您将ID作为作业参数传递，并且您希望在配置步骤时根据该ID从数据库动态获取块大小，因此您可以使用作业范围的步骤，如下所示：

@Bean
@JobScope
public Step sampleStep(@Value("#{jobParameters['id']}") Integer id){
   int chunkSize = myRepo.findChunkSize(id); // this method call fetches chunksize from the db table using the id job parameter
   return myStepBuilderFactory.get("sampleStep")
                .<MyClass, MyClass>chunk(chunkSize)
                .reader(myReader.sampleReader())
                .writer(myWriter.sampleWriter())                        
                .build();
}