使用MultiResourceItemReader时如何为步骤设置动态chunk大小？

核心实现思路 #

Spring Batch的chunk配置不仅支持固定整数值，还支持自定义CompletionPolicy接口实现动态提交逻辑，刚好可以适配你按单个资源行数动态调整chunk大小的需求，不需要改动现有MultiResourceItemReader的基础结构。

具体实现步骤 #

第一步：实现资源行数统计与上下文传递 #

通过MultiResourceItemReader的资源切换监听器，在读取每个资源前先统计该资源的行数，存入Step执行上下文供后续动态判断使用：

public class ResourceChunkSizeListener extends MultiResourceItemReaderListenerSupport<ItemPojo> {
    private StepExecution stepExecution;

    @BeforeStep
    public void beforeStep(StepExecution stepExecution) {
        this.stepExecution = stepExecution;
    }

    // 每次切换到新资源前触发
    @Override
    public void beforeResource(Resource resource) {
        int lineCount = countResourceLines(resource);
        // 将当前资源对应chunk大小存入Step上下文
        stepExecution.getExecutionContext().putInt("currentChunkSize", lineCount);
    }

    // 统计资源行数，可根据业务调整是否排除表头、空行
    private int countResourceLines(Resource resource) {
        try (BufferedReader reader = new BufferedReader(new InputStreamReader(resource.getInputStream()))) {
            int count = 0;
            while (reader.readLine() != null) count++;
            return Math.max(count, 1);
        } catch (IOException e) {
            // 异常时返回默认fallback值
            return 1;
        }
    }
}

第二步：自定义CompletionPolicy实现动态chunk判断 #

public class DynamicChunkCompletionPolicy extends CompletionPolicySupport {
    private int currentItemCount = 0;
    private StepExecution stepExecution;

    @BeforeStep
    public void beforeStep(StepExecution stepExecution) {
        this.stepExecution = stepExecution;
    }

    // 每读取一条Item后判断是否需要提交当前chunk
    @Override
    public boolean isComplete(RepeatContext context, RepeatStatus status) {
        currentItemCount++;
        int chunkSize = stepExecution.getExecutionContext().getInt("currentChunkSize", 3);
        if (currentItemCount >= chunkSize) {
            currentItemCount = 0;
            return true;
        }
        return !status.isContinuable();
    }

    @Override
    public boolean isComplete(RepeatContext context) {
        int chunkSize = stepExecution.getExecutionContext().getInt("currentChunkSize", 3);
        return currentItemCount >= chunkSize;
    }

    // 新chunk启动时重置计数器
    @Override
    public RepeatContext start(RepeatContext parent) {
        currentItemCount = 0;
        return super.start(parent);
    }
}

第三步：修改Step配置替换固定chunk #

@Bean
public Step processStep(){
    return stepBuilderFactory.get("processStep")
            .<itempojo,itempojo>chunk(dynamicChunkCompletionPolicy()) // 替换固定chunk值为自定义策略
            .reader(multiResourceItemReader())
            .processor(processor())
            .writer(writer())
            .listener(readerListener())
            .listener(resourceChunkSizeListener()) // 注册资源切换监听器
            .listener(dynamicChunkCompletionPolicy()) // 注册动态chunk策略监听器
            .build();
}

// 注册对应Bean
@Bean
public DynamicChunkCompletionPolicy dynamicChunkCompletionPolicy(){
    return new DynamicChunkCompletionPolicy();
}

@Bean
public ResourceChunkSizeListener resourceChunkSizeListener(){
    return new ResourceChunkSizeListener();
}

可选替代方案（适合资源数量较少的场景） #

如果你的业务允许每个资源对应独立Step，也可以用动态生成Step的方案，每个Step绑定单个资源，直接设置对应资源行数的固定chunk，逻辑更简单，排查问题更便捷。

注意事项 #

• 大文件场景下提前统计行数会产生额外IO消耗，可在资源接入阶段提前预存行数，避免读取前重复扫描
• 如果业务需要跳过表头、空行，统计行数时要同步调整数值，避免chunk大小与实际有效行数不匹配
• 建议设置合理的fallback chunk值，避免资源读取异常时出现空指针或死循环

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