Spring Data JPA并发插入PostgreSQL重复键约束异常问题

问题本质 #

你当前使用的PESSIMISTIC_WRITE悲观锁仅对已存在的数据库行生效。当并发事务查询不存在的Appliance ID时，没有对应的行可以锁定，所有事务都会判定实体不存在，进而执行插入操作，最终触发appliance_session_id_uq唯一约束冲突。

可行解决方案 #

1. 数据库层面使用UPSERT（推荐） #

利用PostgreSQL原生的INSERT ... ON CONFLICT ... DO UPDATE语法（即UPSERT），直接在数据库层面保证原子性，避免并发插入冲突。

实现方式：

在Repository中定义自定义批量UPSERT方法，替代默认的saveAll：

public interface ApplianceRecordRepository extends JpaRepository<ApplianceRecord, String> {

    @Modifying
    @Transactional
    @Query(value = """
        INSERT INTO appliance_record (id, session_id, column1, column2)
        VALUES (:id, :sessionId, :column1, :column2)
        ON CONFLICT (session_id) DO UPDATE 
        SET column1 = EXCLUDED.column1, column2 = EXCLUDED.column2
        """, nativeQuery = true)
    void upsertSingleAppliance(
        @Param("id") String id,
        @Param("sessionId") String sessionId,
        @Param("column1") String column1,
        @Param("column2") Integer column2
    );

    // 批量UPSERT版本
    @Modifying
    @Transactional
    @Query(value = """
        INSERT INTO appliance_record (id, session_id, column1, column2)
        VALUES (:ids, :sessionIds, :column1s, :column2s)
        ON CONFLICT (session_id) DO UPDATE 
        SET column1 = EXCLUDED.column1, column2 = EXCLUDED.column2
        """, nativeQuery = true)
    void upsertBatchAppliances(
        @Param("ids") List<String> ids,
        @Param("sessionIds") List<String> sessionIds,
        @Param("column1s") List<String> column1s,
        @Param("column2s") List<Integer> column2s
    );
}

优点：

• 数据库层面原子性保证，最可靠；
• 性能最优，无需额外锁或重试；

缺点：

• 依赖PostgreSQL特定语法，耦合数据库；
• 需要手动维护SQL字段，实体字段变更时需同步更新。

2. 应用层加分布式锁 #

针对每个Appliance ID加分布式锁，确保同一时间只有一个事务处理该ID的实体创建/更新。

实现方式（基于数据库锁表）：

1. 先创建锁表：

CREATE TABLE appliance_lock (
    appliance_id VARCHAR(255) PRIMARY KEY,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

2. 在服务方法中先获取锁：

@Transactional
public List<ApplianceRecord> updateAppliances(List<String> applianceIds) {
    // 先获取所有目标ID的锁
    for (String id : applianceIds) {
        // 尝试插入锁记录，已存在则不操作
        repository.insertLock(id);
    }

    try {
        List<ApplianceRecord> existingAppliances = repository.findWithIds(applianceIds);
        List<ApplianceRecord> appliances = service.getAppliances(existingAppliances, applianceIds);
        return repository.saveAll(appliances);
    } finally {
        // 释放锁（可选，也可设置过期时间自动清理）
        repository.deleteLocks(applianceIds);
    }
}

3. Repository中定义锁操作：

@Modifying
@Transactional
@Query(value = "INSERT INTO appliance_lock (appliance_id) VALUES (:id) ON CONFLICT DO NOTHING", nativeQuery = true)
void insertLock(@Param("id") String id);

@Modifying
@Transactional
@Query(value = "DELETE FROM appliance_lock WHERE appliance_id IN (:ids)", nativeQuery = true)
void deleteLocks(@Param("ids") List<String> ids);

优点：

• 不依赖数据库特定语法，适配性强；
• 精准控制并发范围；

缺点：

• 需额外维护锁表或依赖Redis等分布式锁组件；
• 锁的过期和释放逻辑需妥善处理，避免死锁。

3. 捕获异常并重试 #

当触发唯一约束冲突时，重试整个操作流程，利用重试时的查询获取已存在的实体，转为更新操作。

实现方式：

@Transactional
public List<ApplianceRecord> updateAppliances(List<String> applianceIds) {
    int maxRetries = 3;
    int retryCount = 0;

    while (retryCount < maxRetries) {
        try {
            List<ApplianceRecord> existingAppliances = repository.findWithIds(applianceIds);
            List<ApplianceRecord> appliances = service.getAppliances(existingAppliances, applianceIds);
            return repository.saveAll(appliances);
        } catch (ConstraintViolationException e) {
            // 仅处理目标唯一约束的冲突
            if (e.getMessage().contains("appliance_session_id_uq")) {
                retryCount++;
                // 短暂等待，让前一个事务提交
                try {
                    Thread.sleep(100);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    throw new RuntimeException(ie);
                }
            } else {
                // 其他异常直接抛出
                throw e;
            }
        }
    }
    throw new RuntimeException("更新设备失败，已达到最大重试次数");
}

优点：

• 实现简单，无需额外组件或SQL；
• 对现有代码侵入小；

缺点：

• 并发高时会有重试开销；
• 依赖异常捕获，可能掩盖其他问题。

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