如何在 Spring Batch 中将参数从从属步骤传递给阅读器？

Question

我有一个从数据库读取数据的弹簧批处理。基本上发生的事情是我有一个 SQL 查询需要按列（类型）值获取数据。该列有 50 个不同的值。所以有 50 个查询，每个查询都在一个单独的从属步骤上执行。但是查询是在阅读器内部构建的。所以我需要将每种类型传递给 Reader 来构建查询和读取数据。我使用Partitioner 将查询与Offset 和Limit 分开。

这是我的代码，

private Flow flow(List<Step> steps) {
    SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor();
    taskExecutor.setConcurrencyLimit(1);

    return new FlowBuilder<SimpleFlow>("flow")
            .split(taskExecutor).add(steps.stream().map(step -> new FlowBuilder<Flow>("flow_" + step.getName())
                    .start(step).build()).toArray(Flow[]::new)).build();
}

@Bean
public Job job() {
    List<Step> masterSteps = TYPES.stream().map(this::masterStep).collect(Collectors.toList());
    return jobBuilderFactory.get("job")
            .incrementer(new RunIdIncrementer())
            .start(flow(masterSteps))
            .end()
            .build();
}

@Bean
@SneakyThrows
public Step slaveStep(String type) {
    return stepBuilderFactory.get("slaveStep")
            .<User, User>chunk(100)
            .reader(reader(type, 0, 0))
            .writer(writer())
            .build();
}

@Bean
@SneakyThrows
public Step masterStep(String type) {
    return stepBuilderFactory.get("masterStep")
            .partitioner(slaveStep(type).getName(), partitioner(0))
            .step(slaveStep(type))
            .gridSize(5)
            .taskExecutor(executor)
            .build();
}

@Bean
@StepScope
@SneakyThrows
public JdbcCursorItemReader<User> reader(String type,
                                        @Value("#{stepExecutionContext['offset']}") Integer offset,
                                        @Value("#{stepExecutionContext['limit']}") Integer limit) {
    String query = MessageFormat.format(SELECT_QUERY, type, offset, limit); // Ex: SELECT * FROM users WHERE type = 'type' OFFSET 500 LIMIT 1000;

    JdbcCursorItemReader<User> itemReader = new JdbcCursorItemReader<>();
    itemReader.setSql(query);
    itemReader.setDataSource(dataSource);
    itemReader.setRowMapper(new UserMapper());
    itemReader.afterPropertiesSet();

    return itemReader;
}

@Bean
@StepScope
public ItemWriter<User> writer() {
    return new Writer();
}

@Bean
@StepScope
public Partitioner partitioner(@Value("#{jobParameters['limit']}") int limit) {
    return new Partitioner(limit);
}

我使用的问题是reader() 方法type 值没有通过。即使当我添加@Bean 注释时，它也会说Could not autowire. No beans of 'String' type found.。如果我没有输入 @Bean offset 和 limit 总是 0 因为 @Value 没有填充。现在，当我执行批处理时，阅读器内部没有任何反应，因为类型为空。当我对它的工作值进行硬编码时。那么我该如何解决这个问题呢？提前致谢。

Answer 1

如果您要迭代每个TYPE 并执行masterStep，为什么不删除TYPE 逻辑，而是可以SELECT * FROM table OFFSET ? LIMIT ? 并在offset 中处理offset 和limit 内部Partitioner？然后你的 5 个线程将处理这个问题。如果您的最终目标是处理该表中的每条记录，那么您可以简单地使用它而无需担心 TYPE 并在单独的 Step 中执行它们。