Spring Batch并发步骤复用同一ItemReader的实现方案咨询

问题根源分析 #

1. ItemReader状态冲突：默认Spring Bean是单例模式，你的productItemsReader被两个并发步骤共享，而ItemReader通常是有状态的（比如记录当前读取位置、游标位置），多线程并发访问会导致状态混乱，出现数据重复读取、遗漏或异常。
2. 继承方式的配置冲突：你的BaseProcessingStep标注了@Configuration，子类继承后会重复注册同名的productsProcessingStep Bean，触发Spring Bean定义冲突。

最优解决方案 #

1. 调整ItemReader的作用域为原型 #

确保每个步骤获取独立的Reader实例，避免状态共享：

@Bean
@Scope("prototype") // 每次注入都会创建新实例
public ProductItemsReader productItemsReader(DataSource mySqlDataSource) {
    // 示例：构建JdbcCursorItemReader或其他Reader实现
    JdbcCursorItemReader<Product> reader = new JdbcCursorItemReader<>();
    reader.setDataSource(mySqlDataSource);
    reader.setSql("SELECT * FROM products");
    reader.setRowMapper(new BeanPropertyRowMapper<>(Product.class));
    return reader;
}

2. 用组合替代继承复用Step逻辑 #

放弃@Configuration基类继承，改用工具类组合的方式复用Step构建代码，避免Bean注册冲突：

@Component
@RequiredArgsConstructor
public class ProductStepBuilder {

    private final StepBuilderFactory stepBuilderFactory;
    private final ProductItemsReader productItemsReader;

    @Value("${processing.chunk-size}")
    private int chunkSize;

    // 通用Step构建方法，支持传入自定义Processor和Writer
    public Step buildProductStep(String stepName, ItemProcessor<Product, Product> processor, ItemWriter<Product> writer) {
        StepBuilder stepBuilder = stepBuilderFactory.get(stepName)
                .<Product, Product>chunk(chunkSize)
                .reader(productItemsReader); // 原型作用域，每次调用获取新实例

        if (processor != null) {
            stepBuilder = stepBuilder.processor(processor);
        }

        return stepBuilder.writer(writer)
                .build();
    }
}

3. 定义独立的步骤与Job配置 #

分别创建更新、插入步骤，通过组合工具类构建，同时配置并发Flow：

@Configuration
@RequiredArgsConstructor
public class ProductJobConfig {

    private final JobRepository jobRepository;
    private final TaskExecutor taskExecutor;
    private final ProductStepBuilder productStepBuilder;

    // 注入自定义的Writer和Processor
    private final ProductUpdateWriter productUpdateWriter;
    private final ProductInsertWriter productInsertWriter;
    private final ProductDetailsProcessor productDetailsProcessor;

    @Bean("productsUpdateStep")
    public Step productsUpdateStep() {
        // 构建更新步骤，带Processor
        return productStepBuilder.buildProductStep(
                "productsUpdateStep",
                productDetailsProcessor,
                productUpdateWriter
        );
    }

    @Bean("productsInsertStep")
    public Step productsInsertStep() {
        // 构建插入步骤，无需Processor
        return productStepBuilder.buildProductStep(
                "productsInsertStep",
                null,
                productInsertWriter
        );
    }

    @Bean
    public Job productProcessingJob() {
        return new JobBuilder("productProcessingJob", jobRepository)
                .start(splitFlow())
                .build()
                .build();
    }

    @Bean
    public Flow splitFlow() {
        return new FlowBuilder<SimpleFlow>("splitFlow")
                .split(taskExecutor)
                .add(updateFlow(), insertFlow())
                .build();
    }

    @Bean
    public Flow updateFlow() {
        return new FlowBuilder<SimpleFlow>("updateFlow")
                .start(productsUpdateStep())
                .build();
    }

    @Bean
    public Flow insertFlow() {
        return new FlowBuilder<SimpleFlow>("insertFlow")
                .start(productsInsertStep())
                .build();
    }
}

关键说明 #

• 原型Reader的必要性：每个并发步骤必须拥有独立的Reader实例，确保各自的读取状态互不干扰，避免数据一致性问题。
• 组合优于继承：组合方式更灵活，避免了@Configuration继承导致的Bean名称冲突，同时符合面向对象的依赖倒置原则。
• 步骤独立性：每个步骤使用唯一的Bean名称，Spring可以正确识别并管理，并发执行时不会出现资源竞争。

内容的提问来源于stack exchange，提问作者BreenDeen