Spring Batch中如何在写入所有Chunk后仅执行一次事务提交？

针对你提出的需求——Chunk式处理避免OOM、任务完成后一次性提交事务、无需中间表/文件、支持重启补写，以下是符合Spring Batch最佳实践的实现建议：

核心思路说明 #

全局事务与重启时的部分进度确认存在天然矛盾：全局事务要么全提交要么全回滚，中途失败时不会留下已处理的持久化记录。因此我们需要结合可重启的读取器跟踪处理进度，延迟提交的自定义Writer实现全局事务，同时通过幂等写入避免重启后重复处理的问题。

一、配置可重启的读取器 #

必须使用支持重启的ItemReader，确保任务重启时能从上次中断的位置继续读取，避免重复读取已处理条目。推荐使用JdbcPagingItemReader，它会将分页位置（如最后读取的ID、当前页号）存储到Spring Batch的ExecutionContext中。

@Bean
public JdbcPagingItemReader<SourceEntity> sourceReader(DataSource dataSource) {
    return new JdbcPagingItemReaderBuilder<SourceEntity>()
            .dataSource(dataSource)
            .name("sourceReader")
            .selectClause("SELECT id, column1, column2")
            .fromClause("FROM source_table")
            .sortKeys(Map.of("id", Order.ASCENDING)) // 必须指定排序键，用于分页和重启定位
            .rowMapper(new BeanPropertyRowMapper<>(SourceEntity.class))
            .build();
}

二、实现延迟提交的自定义ItemWriter #

自定义ItemWriter并实现StepExecutionListener，在每个Chunk处理时缓存批量写入操作，直到Step执行完成后一次性提交事务。这种方式既保留Chunk处理的内存优势，又实现全局事务提交。

public class DeferredCommitJdbcBatchItemWriter<T> implements ItemWriter<T>, StepExecutionListener {

    private final DataSource dataSource;
    private final String writeSql;
    private final ItemPreparedStatementSetter<T> statementSetter;

    private Connection connection;
    private PreparedStatement preparedStatement;
    private int batchCounter;

    public DeferredCommitJdbcBatchItemWriter(DataSource dataSource, String writeSql,
                                            ItemPreparedStatementSetter<T> statementSetter) {
        this.dataSource = dataSource;
        this.writeSql = writeSql;
        this.statementSetter = statementSetter;
    }

    @Override
    public void beforeStep(StepExecution stepExecution) {
        try {
            // 获取连接并禁用自动提交，开启全局事务
            connection = dataSource.getConnection();
            connection.setAutoCommit(false);
            preparedStatement = connection.prepareStatement(writeSql);
        } catch (SQLException e) {
            throw new BatchRuntimeException("初始化数据库连接失败", e);
        }
    }

    @Override
    public void write(List<? extends T> items) throws Exception {
        // 将当前Chunk的条目添加到批量操作缓存，暂不执行
        for (T item : items) {
            statementSetter.setValues(item, preparedStatement);
            preparedStatement.addBatch();
            batchCounter++;
        }
    }

    @Override
    public ExitStatus afterStep(StepExecution stepExecution) {
        try {
            if (batchCounter > 0) {
                // 执行所有缓存的批量操作并提交全局事务
                preparedStatement.executeBatch();
                connection.commit();
            }
            return ExitStatus.COMPLETED;
        } catch (SQLException e) {
            try {
                connection.rollback();
            } catch (SQLException ex) {
                throw new BatchRuntimeException("事务回滚失败", ex);
            }
            throw new BatchRuntimeException("批量提交失败", e);
        } finally {
            // 关闭资源
            try {
                if (preparedStatement != null) preparedStatement.close();
                if (connection != null) connection.close();
            } catch (SQLException e) {
                throw new BatchRuntimeException("资源关闭失败", e);
            }
        }
    }
}

三、配置Step的全局事务 #

需要调整Step的事务属性，禁用默认的Chunk级事务提交，让整个Step运行在一个全局事务中，同时注册自定义Writer作为Step监听器。

@Bean
public Step dataMigrationStep(JobRepository jobRepository,
                              PlatformTransactionManager transactionManager,
                              ItemReader<SourceEntity> sourceReader,
                              ItemProcessor<SourceEntity, TargetEntity> processor,
                              DeferredCommitJdbcBatchItemWriter<TargetEntity> writer) {
    // 配置全局事务属性
    DefaultTransactionAttribute transactionAttribute = new DefaultTransactionAttribute();
    transactionAttribute.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED);
    transactionAttribute.setTimeout(3600); // 根据业务调整超时时间

    return new StepBuilder("dataMigrationStep", jobRepository)
            .<SourceEntity, TargetEntity>chunk(1000, transactionManager)
            .reader(sourceReader)
            .processor(processor)
            .writer(writer)
            .transactionAttribute(transactionAttribute)
            .listener(writer) // 注册Step监听器，触发全局提交/回滚
            .build();
}

四、关键注意事项 #

1. OOM风险规避：如果数据集极大，缓存所有批量操作参数可能占用过多内存。可调整Chunk大小，或采用数据库端的批量缓存机制（如MySQL的rewriteBatchedStatements参数优化批量写入）。
2. 幂等性保障：由于全局事务失败后重启会重新处理已读取的条目，目标表必须通过唯一主键或业务唯一键实现幂等写入（如使用INSERT ... ON DUPLICATE KEY UPDATE或MERGE语句）。
3. 事务超时配置：全局事务持续时间较长，需调整数据库和事务管理器的超时时间，避免因超时导致任务失败。
4. 元数据独立性：Spring Batch的Step执行进度（如已读取条目数）会被独立持久化到元数据表中，即使业务事务未提交，重启时仍能从正确位置继续读取。

五、推荐替代方案：Chunk事务+幂等写入（最佳实践） #

如果可以接受Chunk级事务提交，这是Spring Batch官方推荐的最优方案，兼顾性能、可靠性和可维护性：

• 使用默认的JdbcBatchItemWriter，每个Chunk自动提交事务。
• 目标表设置唯一约束，写入时使用幂等SQL语句（如INSERT ... ON DUPLICATE KEY UPDATE）。
• 读取器使用可重启实现，重启时从上次位置继续读取，重复条目会被幂等逻辑自动忽略。

这种方案无需自定义Writer，避免了全局事务的锁竞争、超时等风险，同时完全满足你的需求。

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