Spring Boot Batch实现ItemWriter多线程并行写入数据库的方案问询

问题原因 #

1. 手动实例化JpaItemWriter未初始化必要依赖：Spring自动注入的JpaItemWriter已经默认加载了你的EntityManagerFactory并完成初始化，你手动new的实例没有调用setEntityManagerFactory()方法注入依赖，所以直接抛出缺少EntityManagerFactory的异常。
2. CompositeItemWriter本身不支持并行执行：它的默认逻辑是串行依次调用所有委托写入器，你哪怕配置10个委托，也还是串行执行，完全达不到并行写入的预期。

最佳实现方案 #

方案1：AsyncItemWriter异步包装（轻量场景，代码改动最小） #

适合数据量中等，不需要拆分读操作的场景，只需要把现有写入器包装成异步执行即可：
首先配置业务专用线程池：

@Bean
public TaskExecutor batchWriteTaskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    // 核心线程数根据你的数据库连接池最大连接数调整，不要超过DB连接池上限
    executor.setCorePoolSize(10);
    executor.setMaxPoolSize(10);
    executor.setQueueCapacity(100);
    executor.setThreadNamePrefix("batch-write-thread-");
    executor.initialize();
    return executor;
}

然后修改写入器配置，用AsyncItemWriter包装原有JpaItemWriter：

@Bean
public AsyncItemWriter<SoftLayerData> asyncSoftlayerDataWriter(
    JpaItemWriter<SoftLayerData> softlayerDataWriter,
    TaskExecutor batchWriteTaskExecutor) {
    AsyncItemWriter<SoftLayerData> asyncWriter = new AsyncItemWriter<>();
    asyncWriter.setDelegate(softlayerDataWriter);
    asyncWriter.setTaskExecutor(batchWriteTaskExecutor);
    return asyncWriter;
}
// 原有CompositeItemWriter如果没有其他委托逻辑可以直接删除，Step直接配置使用asyncWriter即可

注意：需要同步调整你的Step配置，设置合适的chunk大小，同时确保数据库连接池的最大连接数大于等于异步写入的线程数，避免线程阻塞等待连接。

方案2：分区Step（大数据量场景，读写全并行） #

如果数据量极大，连读取操作都有性能瓶颈，可以使用分区Step，将整个数据集拆分为多个独立分区，每个分区由独立的线程完成读、处理、写全流程：

// 配置分区器，根据你的数据范围拆分分区
@Bean
public Partitioner softlayerDataPartitioner() {
    return gridSize -> {
        Map<String, ExecutionContext> partitions = new HashMap<>();
        // 示例：按行数拆分10个分区，每个分区处理10000行
        for (int i = 0; i < 10; i++) {
            ExecutionContext context = new ExecutionContext();
            context.putInt("startRow", i * 10000);
            context.putInt("endRow", (i + 1) * 10000);
            partitions.put("partition_" + i, context);
        }
        return partitions;
    };
}

// 配置从Step，每个分区执行的逻辑
@Bean
public Step softlayerSlaveStep(
    StepBuilderFactory stepBuilderFactory,
    ItemReader<SoftLayerData> softLayerReader,
    ItemWriter<SoftLayerData> softLayerWriter) {
    return stepBuilderFactory.get("softlayerSlaveStep")
        .<SoftLayerData, SoftLayerData>chunk(1000)
        .reader(softLayerReader)
        .writer(softLayerWriter)
        .build();
}

// 配置主Step，调度分区执行
@Bean
public Step softlayerMasterStep(
    StepBuilderFactory stepBuilderFactory,
    Partitioner softlayerDataPartitioner,
    Step softlayerSlaveStep,
    TaskExecutor batchWriteTaskExecutor) {
    return stepBuilderFactory.get("softlayerMasterStep")
        .partitioner(softlayerSlaveStep.getName(), softlayerDataPartitioner)
        .step(softlayerSlaveStep)
        .taskExecutor(batchWriteTaskExecutor)
        .gridSize(10) // 和分区数一致
        .build();
}

注意：分区模式下你的读取器需要支持按ExecutionContext传入的参数读取对应分区的数据，避免重复读取。

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