如何确保Hibernate批量删除与保存操作的正确性

核心现象 #

定时任务通过JPA Repository同步第三方数据至自有库，流程为：筛选未迁移且非空的外部数据ID→分批处理→转换数据→按外键删除原有内部数据→保存新数据→标记外部数据为已迁移。处理约2000条数据后，内部表部分数据丢失，但所有外部数据被标记为已迁移，使用MSSQL数据库，已配置hibernate.order_inserts和hibernate.order_updates。

可能根因 #

1. 大事务边界导致的状态不一致
   原run()方法使用全局事务包裹所有批次，EntityManager一级缓存会随数据量膨胀，导致实体状态追踪失效；若某批次出现隐性异常（如约束冲突、连接超时），可能出现部分数据提交、部分回滚的不一致情况。

2. 删除操作的低效与缓存问题
   通过getAllByForeignKeyIn加载实体再执行deleteAll，不仅效率低下，还可能因JPA缓存未及时刷新，导致删除操作未真正生效，后续插入被覆盖或丢失。

3. 迁移标记逻辑的漏洞
   仅校验InternalEntityOne的保存结果就标记外部数据为已迁移，忽略了InternalEntityTwo的保存状态；若InternalEntityTwo保存失败但InternalEntityOne成功，会出现外部数据标记已迁移，但对应内部数据缺失的情况。

4. JPA批量配置缺失
   仅开启排序配置但未设置hibernate.jdbc.batch_size，导致JPA未真正执行批量SQL，单条执行大量数据时易出现静默失败。

具体解决方案 #

1. 拆分事务边界，缩小事务范围 #

移除run()方法的全局事务，为每个批次创建独立事务，避免缓存膨胀与跨批次影响：

@Override
public void run() {
    List<Long> idsMatchingCriteria = externalDataJpaRepo.findAllIdsByMigratedIsFalseAndContentIsNotNull();
    List<List<Long>> idsBatches = idsMatchingCriteria.size() < BATCH_SIZE 
        ? List.of(idsMatchingCriteria) 
        : ListUtils.partition(idsMatchingCriteria, BATCH_SIZE);
    
    for (List<Long> list : idsBatches) {
        transactionUtil.withNewTransaction(() -> {
            List<ExternalEntity> externalEntitiesToProcess = externalDataJpaRepo.findAllById(list);
            List<Long> savedForeignIds = persistBatchOfTransformedEntities(externalEntitiesToProcess);
            
            // 校验所有关联外键都完成同步
            Set<Long> targetForeignIds = externalEntitiesToProcess.stream()
                .map(ExternalEntity::getForeignId)
                .collect(Collectors.toSet());
            if (targetForeignIds.equals(new HashSet<>(savedForeignIds))) {
                externalEntitiesToProcess.forEach(cj -> cj.setMigrated(true));
                externalDataJpaRepo.saveAll(externalEntitiesToProcess);
            } else {
                log.error("Batch {} incomplete: some foreign ids not saved", list);
                throw new RuntimeException("Batch processing failed, rollback");
            }
        });
    }
}

2. 优化删除操作，使用批量JPQL #

替换加载实体再删除的逻辑，直接通过JPQL批量删除，避免缓存干扰：
在InternalRepoTwo中添加批量删除方法：

@Modifying
@Query("DELETE FROM InternalEntityTwo e WHERE e.foreignId IN :foreignIds")
void deleteByForeignIdIn(@Param("foreignIds") List<Long> foreignIds);

修改persistBatchOfTransformedEntities中的删除逻辑：

// 替换原加载实体删除的代码
internalRepoTwo.deleteByForeignIdIn(foreignIdsList);
internalRepoTwo.flush(); // 确保删除立即生效

3. 完善数据一致性校验 #

同时校验InternalEntityOne和InternalEntityTwo的保存结果，确保所有关联数据同步完成：

List<Long> persistBatchOfTransformedEntities(List<ExternalEntity> externalEntities) {
    // 转换逻辑...
    Set<Long> targetForeignIds = externalEntities.stream()
        .map(ExternalEntity::getForeignId)
        .collect(Collectors.toSet());

    // 处理InternalEntityOne
    List<InternalEntityOne> savedInternalOne = persistInternalEntitiesOne(internalEntityOneToUpdate, transformedInternalEntitiesOne);
    Set<Long> savedOneForeignIds = savedInternalOne.stream()
        .map(InternalEntityOne::getForeignId)
        .collect(Collectors.toSet());

    // 处理InternalEntityTwo
    internalRepoTwo.deleteByForeignIdIn(new ArrayList<>(targetForeignIds));
    internalRepoTwo.flush();
    List<InternalEntityTwo> savedInternalTwo = internalRepoTwo.saveAllAndFlush(transformedInterlanEntitiesTwo);
    Set<Long> savedTwoForeignIds = savedInternalTwo.stream()
        .map(InternalEntityTwo::getForeignId)
        .collect(Collectors.toSet());

    // 双重校验
    if (!targetForeignIds.equals(savedOneForeignIds) || !targetForeignIds.equals(savedTwoForeignIds)) {
        throw new RuntimeException("Failed to save all internal entities for foreign ids: " + targetForeignIds);
    }

    return new ArrayList<>(targetForeignIds);
}

4. 补充JPA批量操作配置 #

在Hibernate配置中添加批量相关属性，确保批量SQL生效：

private Properties additionalProperties() {
    Properties properties = new Properties();
    properties.setProperty("hibernate.order_inserts", "true");
    properties.setProperty("hibernate.order_updates", "true");
    properties.setProperty("hibernate.jdbc.batch_size", "50"); // 与批次大小一致
    properties.setProperty("hibernate.jdbc.batch_versioned_data", "true"); // 适配MSSQL批量更新
    properties.setProperty("hibernate.format_sql", "true"); // 可选：开启SQL格式化便于调试
    return properties;
}

5. 增强日志与异常追踪 #

在关键步骤添加日志，记录批次处理数量、SQL执行结果；捕获并记录异常，避免静默失败：

// 在persistBatchOfTransformedEntities中添加
log.info("Processing {} external entities, target foreign ids: {}", externalEntities.size(), targetForeignIds);
log.info("Deleted {} InternalEntityTwo records", targetForeignIds.size());
log.info("Saved {} InternalEntityOne, {} InternalEntityTwo records", savedInternalOne.size(), savedInternalTwo.size());

验证步骤 #

1. 先测试小批量数据（如10条），确认数据一致性
2. 逐步增大批次规模，观察是否出现数据丢失
3. 开启SQL日志，检查生成的删除、插入SQL是否符合预期
4. 模拟异常场景（如约束违反），验证事务回滚逻辑与迁移标记的正确性

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