使用Spring Batch的ThreadPoolExecutor无法限制并行度问题排查

我的配置 #

@StepScope
@Bean(name = "mySlaveStep")
public Step mySlaveStep(
        @Qualifier(value = "myReader") ItemReader reader,
        @Qualifier(value = "myWriter") ItemWriter writer,
        StepBuilderFactory stepBuilderFactory) {
    return stepBuilderFactory.get("MySlaveStep")
            .<SomeObject, SomeObject>chunk(1000)
            .reader(reader)
            .writer(writer)
            .build();
}

@Bean(name = "myStep")
public Step myStep(
        @Qualifier(value = "myPartitioner") Partitioner partitioner, // 带有@StepScope
        @Qualifier(value = "myExecutor") TaskExecutor executor, // 加不加@StepScope都试过
        @Qualifier(value = "myStep") Step step, // 带有@StepScope —— 此处存在循环依赖错误
        StepBuilderFactory stepBuilderFactory) {
    return stepBuilderFactory
            .get("MyStep")
            .partitioner("MyPartition", partitioner)
            .taskExecutor(executor)
            .step(step)
            .build();
}

@StepScope // 加不加都试过
@Bean(name = "taskExecutor")
public TaskExecutor taskExecutor() {
    final ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setQueueCapacity(Integer.MAX_VALUE);
    executor.setCorePoolSize(2);
    executor.setMaxPoolSize(2);
    return executor;
}

异常信息 #

Caused by: org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'scopedTarget.mySlaveStep': Scope 'step' is not active for the current thread; consider defining a scoped proxy for this bean if you intend to refer to it from a singleton; nested exception is java.lang.IllegalStateException: No context holder available for step scope
    at org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:368) ~[spring-beans-5.2.0.RELEASE.jar:5.2.0.RELEASE]
    at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:202) ~[spring-beans-5.2.0.RELEASE.jar:5.2.0.RELEASE]
    at org.springframework.aop.target.SimpleBeanTargetSource.getTarget(SimpleBeanTargetSource.java:35) ~[spring-aop-5.2.0.RELEASE.jar:5.2.0.RELEASE]
    at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:192) ~[spring-aop-5.2.0.RELEASE.jar:5.2.0.RELEASE]
    at com.sun.proxy.$Proxy144.execute(Unknown Source) ~[na:na]
    at org.springframework.batch.core.partition.support.TaskExecutorPartitionHandler$1.call(TaskExecutorPartitionHandler.java:138) ~[spring-batch-core-4.2.0.RELEASE.jar:4.2.0.RELEASE]
    at org.springframework.batch.core.partition.support.TaskExecutorPartitionHandler$1.call(TaskExecutorPartitionHandler.java:135) ~[spring-batch-core-4.2.0.RELEASE.jar:4.2.0.RELEASE]
    at java.base/java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:264) ~[na:na]
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java) ~[na:na]
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) ~[na:na]
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) ~[na:na]
    at java.base/java.lang.Thread.run(Thread.java:829) ~[na:na]
Caused by: java.lang.IllegalStateException: No context holder available for step scope
    at org.springframework.batch.core.scope.StepScope.getContext(StepScope.java:167) ~[spring-batch-core-4.2.0.RELEASE.jar:4.2.0.RELEASE]
    at org.springframework.batch.core.scope.StepScope.get(StepScope.java:99) ~[spring-batch-core-4.2.0.RELEASE.jar:4.2.0.RELEASE]
    at org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:356) ~[spring-beans-5.2.0.RELEASE.jar:5.2.0.RELEASE]

问题说明 #

我尝试通过自定义TaskExecutor限制分区任务的并行线程数，但无论给TaskExecutor加不加@StepScope注解，都会抛出上述StepScope异常；只有完全移除TaskExecutor改用默认实现时，问题才会消失。我需要实现限制并行分区数量的同时避免这个异常。

解决方案 #

1. 修复循环依赖错误 #

你的myStep方法中存在明显的循环依赖：参数里引用了@Qualifier("myStep")的Step bean，这会导致Spring无法正确创建bean。需要将其改为引用你的slave step，即@Qualifier("mySlaveStep")：

@Bean(name = "myStep")
public Step myStep(
        @Qualifier(value = "myPartitioner") Partitioner partitioner,
        @Qualifier(value = "stepContextTaskExecutor") TaskExecutor executor,
        @Qualifier(value = "mySlaveStep") Step slaveStep, // 修正为引用mySlaveStep
        StepBuilderFactory stepBuilderFactory) {
    return stepBuilderFactory
            .get("MyStep")
            .partitioner("MyPartition", partitioner)
            .taskExecutor(executor)
            .step(slaveStep)
            .build();
}

2. 使用StepContextTaskExecutor传递上下文 #

异常的核心原因是：自定义TaskExecutor的子线程无法访问主线程的StepScope上下文。Spring Batch提供了StepContextTaskExecutor，它会自动将主线程的StepContext绑定到子线程，执行完成后再移除，确保StepScope bean能正常创建。

修改TaskExecutor的配置，去掉@StepScope，改为单例bean，并包装成StepContextTaskExecutor：

// 定义普通的线程池TaskExecutor，单例无需@StepScope
@Bean(name = "taskExecutor")
public TaskExecutor taskExecutor() {
    final ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setQueueCapacity(Integer.MAX_VALUE);
    executor.setCorePoolSize(2);
    executor.setMaxPoolSize(2);
    executor.initialize(); // 必须调用initialize方法初始化线程池
    return executor;
}

// 用StepContextTaskExecutor包装上面的线程池，确保上下文传递
@Bean(name = "stepContextTaskExecutor")
public StepContextTaskExecutor stepContextTaskExecutor(@Qualifier("taskExecutor") TaskExecutor delegate) {
    StepContextTaskExecutor stepContextTaskExecutor = new StepContextTaskExecutor();
    stepContextTaskExecutor.setDelegate(delegate);
    return stepContextTaskExecutor;
}

3. 验证效果 #

完成上述修改后，分区任务会使用带有上下文传递能力的StepContextTaskExecutor，既可以限制并行线程数为2，又能确保StepScope的mySlaveStep能在子线程中正常初始化，不会再抛出"No context holder available for step scope"异常。

内容的提问来源于stack exchange，提问作者Hasan Can Saral