Spring Batch分区改造后ItemProcessor/Writer中@BeforeStep未触发

问题描述 #

将Spring Batch实现迁移为分区模式后，发现ItemProcessor和ItemWriter中定义的@BeforeStep注解方法不再被调用，但处理器的process方法与写入器的write方法仍能正常执行。相关核心代码如下：

配置类代码 #

@Bean
@StepScope
public ItemWriter<EmployeeTaxDetail> customerItemWriter()
{
    ItemWriter<EmployeeTaxDetail> itemWriter = appContext.getBean(EmployeeTaxCalculationWriter.class);
    return itemWriter;
}
@Bean
@StepScope
public ItemProcessor<EmployeeDetail, EmployeeTaxDetail> processor() {
    ItemProcessor<EmployeeDetail, EmployeeTaxDetail> itemProcessor = appContext.getBean(EmployeeTaxCalculationProcessor.class);
    return itemProcessor;
}

// Master Step
@Bean
public Step step1() 
{
    return new StepBuilder("step1", jobRepository)
            .partitioner(slaveStep().getName(), partitioner())
            .listener(processor())
            .listener(customerItemWriter())
            .step(slaveStep())
            .gridSize(5)
            .taskExecutor(new SimpleAsyncTaskExecutor())
            .build();
}

// Slave Step
@Bean
public Step slaveStep() 
{
    return new StepBuilder("slaveStep", jobRepository)
            .<EmployeeDetail, EmployeeTaxDetail>chunk(10,transactionManager)
            .reader(pagingItemReader(null,null))
            .processor(processor())
            .writer(customerItemWriter())
            .build();
}

抽象类核心代码 #

AbstractItemWriter中的@BeforeStep方法：

@BeforeStep
public void beforeStep(final StepExecution stepExecution) {
    parameters = stepExecution.getJobExecution().getJobParameters();
}

AbstractItemProcessor中的@BeforeStep方法：

@BeforeStep
public void beforeStep(final StepExecution stepExecution) {
    parameters = stepExecution.getJobExecution().getJobParameters();
}

问题原因 #

1. 分区模式的Step职责划分：
   分区模式下，master step仅负责任务分发与协调，不会执行实际的chunk处理逻辑；真正执行read/process/write流程的是slave step，且每个slave step在独立线程中运行。
   当前代码将processor和writer注册为master step的监听器，但master step不会触发chunk处理阶段的监听器方法，而slave step未将这两个组件注册为自身监听器，导致@BeforeStep无法被触发。

2. StepScope Bean的错误使用：
   定义的processor()和customerItemWriter()标注了@StepScope，但内部直接从Spring根上下文获取单例的组件实例，完全绕过了StepScope的生命周期管理，既无法保证每个slave step拥有独立的组件实例，也干扰了Spring Batch对监听器的识别逻辑。

解决方案 #

方案1：将Processor/Writer注册为Slave Step的监听器 #

修改slaveStep的定义，将processor和writer注册为slave step的监听器，确保slave step执行时触发@BeforeStep方法；同时移除master step中不必要的监听器注册：

// 修改后的Slave Step
@Bean
public Step slaveStep() {
    return new StepBuilder("slaveStep", jobRepository)
            .<EmployeeDetail, EmployeeTaxDetail>chunk(10, transactionManager)
            .reader(pagingItemReader(null,null))
            .processor(processor())
            .writer(customerItemWriter())
            // 添加监听器注册
            .listener(processor())
            .listener(customerItemWriter())
            .build();
}

// 修改后的Master Step
@Bean
public Step step1() {
    return new StepBuilder("step1", jobRepository)
            .partitioner(slaveStep().getName(), partitioner())
            // 移除master step上的监听器注册
            // .listener(processor())
            // .listener(customerItemWriter())
            .step(slaveStep())
            .gridSize(5)
            .taskExecutor(new SimpleAsyncTaskExecutor())
            .build();
}

方案2：修正StepScope Bean的定义（优化实例管理） #

当前的StepScope Bean定义未生效，需调整为直接创建StepScope的组件实例，确保每个slave step能获取到独立的组件实例：

@Bean
@StepScope
public ItemProcessor<EmployeeDetail, EmployeeTaxDetail> processor() {
    // 直接实例化StepScope的处理器
    return new EmployeeTaxCalculationProcessor();
}

@Bean
@StepScope
public ItemWriter<EmployeeTaxDetail> customerItemWriter() {
    // 若需要依赖注入，可通过构造注入传入
    return new EmployeeTaxCalculationWriter();
}

注：如果EmployeeTaxCalculationWriter需要依赖FEBPDataSourceProvider，建议改用构造注入的方式传入，避免依赖Spring上下文的自动注入。

方案3：改用StepExecutionListener接口（替代注解方式） #

若注解方式仍存在识别问题，可让抽象类实现StepExecutionListener接口，显式定义监听器方法，提升Spring Batch对监听器的识别度：

public abstract class AbstractItemWriter<T> implements ItemWriter<T>, StepExecutionListener {

    private JobParameters parameters;

    @Override
    public void write(@NonNull Chunk<? extends T> chunk) throws Exception {
        // 原有write逻辑不变
    }

    protected abstract void executeWrite(@NonNull Chunk<? extends T> chunk) throws Exception;

    @Override
    public void beforeStep(final StepExecution stepExecution) {
        parameters = stepExecution.getJobExecution().getJobParameters();
    }

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

    public JobParameters getParameters() {
        return parameters;
    }
}

同样修改AbstractItemProcessor实现StepExecutionListener接口，再按照方案1的方式注册为slave step的监听器即可。

内容的提问来源于stack exchange，提问作者Shreyas Holla P