Spring Batch是否适用于百万级数据关联处理及事务回滚场景？

你的十万级数据迁移场景完全适合用Spring Batch的Chunk模型，以下是针对你遇到的两个问题的具体解决方案：

问题1：自定义Reader搭配ScrollableResults时Session提前关闭 #

原因 #

Spring Batch的Step事务默认以Chunk为单位，自定义Reader如果未正确绑定Session到Chunk事务生命周期，会导致Session在首次读取后被Spring容器关闭。

解决方案 #

优先使用Spring Batch官方提供的HibernateCursorItemReader，它会自动管理Session生命周期，保持Session在Chunk处理周期内打开，且支持游标式读取（适合大数据量）：

@Bean
public HibernateCursorItemReader<SourceEntity> sourceEntityReader(SessionFactory sessionFactory) {
    HibernateCursorItemReader<SourceEntity> reader = new HibernateCursorItemReader<>();
    reader.setSessionFactory(sessionFactory);
    // 替换为你的关联查询HQL，如需关联其他表直接在HQL中join即可
    reader.setQueryString("SELECT s FROM SourceEntity s LEFT JOIN FETCH s.relatedEntity r");
    reader.setFetchSize(1000); // 设置合适的Fetch Size，减少数据库往返次数，优化性能
    return reader;
}

如果必须自定义Reader，确保使用SessionFactory.getCurrentSession()获取Session（而非openSession()），该Session会自动绑定到当前Chunk的事务中，避免提前关闭：

public class CustomHibernateReader implements ItemReader<SourceEntity> {
    private final SessionFactory sessionFactory;
    private ScrollableResults results;

    @Override
    public SourceEntity read() throws Exception {
        if (results == null) {
            Session session = sessionFactory.getCurrentSession();
            Query query = session.createQuery("SELECT s FROM SourceEntity s");
            query.setFetchSize(1000);
            results = query.scroll(ScrollMode.FORWARD_ONLY);
        }
        return results.next() ? (SourceEntity) results.get(0) : null;
    }
}

问题2：出错时需手动删除已写入数据（事务自动提交） #

原因 #

默认Chunk模型下，每个Chunk执行成功后会自动提交事务，导致部分数据已持久化，出错时无法自动回滚所有已写入数据。

解决方案 #

根据你的数据库性能，选择以下两种方案：

1. 全局事务方案（适合数据库能支撑大事务的场景）
   将整个Step纳入一个全局事务中，Chunk仅作为数据分批处理的单位，事务在整个Step执行完成后才提交，任何异常都会触发全量回滚：

   @Bean
   public Step migrationStep(ItemReader<SourceEntity> reader, 
                             ItemProcessor<SourceEntity, TargetEntity> processor,
                             ItemWriter<TargetEntity> writer,
                             PlatformTransactionManager transactionManager) {
       DefaultTransactionAttribute globalTxAttribute = new DefaultTransactionAttribute();
       globalTxAttribute.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED);
       globalTxAttribute.setTimeout(3600); // 延长事务超时时间，适配大数据量处理
   
       return stepBuilderFactory.get("migrationStep")
               .<SourceEntity, TargetEntity>chunk(1000)
               .reader(reader)
               .processor(processor)
               .writer(writer)
               .transactionManager(transactionManager)
               .transactionAttribute(globalTxAttribute)
               .faultTolerant()
               .rollbackFor(Exception.class) // 所有异常均触发回滚
               .build();
   }
   

   注意：大事务可能导致数据库事务日志膨胀，需提前调整数据库参数（如MySQL的innodb_log_file_size）。

2. 临时表补偿方案（适合无法支撑大事务的场景）
   先将所有处理后的数据写入临时表，待整个Job执行成功后，再批量同步到目标表；若Job失败，直接清空临时表即可：

   • 配置ItemWriter写入临时表（如target_temp）
   • 新增一个afterJob的Listener，在Job成功时执行同步逻辑，失败时清空临时表：

   @Bean
   public JobExecutionListener tempTableCleanupListener(JdbcTemplate jdbcTemplate) {
       return new JobExecutionListener() {
           @Override
           public void beforeJob(JobExecution jobExecution) {
               // 初始化临时表（如不存在则创建）
               jdbcTemplate.execute("CREATE TABLE IF NOT EXISTS target_temp LIKE target_table");
           }
   
           @Override
           public void afterJob(JobExecution jobExecution) {
               if (jobExecution.getStatus() == BatchStatus.COMPLETED) {
                   // 同步临时表到目标表
                   jdbcTemplate.execute("INSERT INTO target_table SELECT * FROM target_temp");
               }
               // 无论成败，最后清空临时表
               jdbcTemplate.execute("TRUNCATE TABLE target_temp");
           }
       };
   }
   

总结 #

Spring Batch的Chunk模型完全适配你的数据迁移场景，无需更换开发方案。通过上述方案解决Session管理和事务回滚问题后，即可高效、安全地完成十万级数据的迁移操作。

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