Spring事务场景下无数据库约束的多线程重复数据插入解决策略问询

你的核心问题是**“检查-插入”操作不具备原子性**导致的竞态条件：多线程并发调用时，多个线程都能通过“邮箱是否存在”的检查，最终重复插入同一用户。以下是在不修改数据库架构（不加唯一约束）、保留事务的前提下的几种可行方案：

1. 悲观锁：查询时加排他锁 #

通过在查询邮箱时添加数据库排他锁，确保同一时间只有一个线程能执行后续插入操作，其他线程会被阻塞直到当前事务完成。

实现步骤： #

• 在UserRepository中添加带锁的查询方法：

public interface UserRepository extends JpaRepository<User, Long> {
    @Lock(LockModeType.PESSIMISTIC_WRITE)
    User findByEmail(String email);
}

• 原Service方法直接使用该带锁查询即可：

@Transactional
public void verifyAndInsertUser(User user) {
    User existingUser = userRepository.findByEmail(user.getEmail());

    if (existingUser == null) {
        userRepository.save(user);
    } else {
        throw new UserExistsException("User with email " + user.getEmail() + " already exists.");
    }
}

注意：这种方式会阻塞其他线程对同一邮箱的查询，适合并发量中等的场景；并发过高时可能导致数据库连接池耗尽。

2. 原子SQL操作：将“检查+插入”合并为单条SQL #

利用数据库的原子性特性，把检查和插入逻辑写成一条SQL，让数据库层面保证操作的原子性，从根源避免竞态条件。

实现步骤： #

• 在UserRepository中定义自定义插入方法：

public interface UserRepository extends JpaRepository<User, Long> {
    @Modifying
    @Transactional
    @Query(value = "INSERT INTO user (email, username, password) " +
                   "SELECT :email, :username, :password " +
                   "WHERE NOT EXISTS (SELECT 1 FROM user WHERE email = :email)",
           nativeQuery = true)
    int saveIfNotExists(@Param("email") String email,
                        @Param("username") String username,
                        @Param("password") String password);
}

• 修改Service方法，根据插入行数判断操作结果：

@Transactional
public void verifyAndInsertUser(User user) {
    int insertedRows = userRepository.saveIfNotExists(
        user.getEmail(),
        user.getUsername(),
        user.getPassword()
    );

    if (insertedRows == 0) {
        throw new UserExistsException("User with email " + user.getEmail() + " already exists.");
    }
}

优势：性能优于悲观锁，因为不需要长时间持有锁，适合大部分高并发场景，主流关系型数据库均支持INSERT ... SELECT语法。

3. 应用层锁：本地锁/分布式锁 #

如果是单实例部署，可以用Java本地锁（如synchronized或ReentrantLock）；如果是多实例集群部署，则需要用分布式锁（如Redis Redisson、ZooKeeper），锁的粒度要精确到用户邮箱，避免全局锁影响性能。

单实例本地锁示例： #

@Service
public class UserService {
    @Autowired
    private UserRepository userRepository;
    // 用ConcurrentHashMap存储每个邮箱对应的锁，避免全局锁
    private final ConcurrentHashMap<String, Object> lockMap = new ConcurrentHashMap<>();

    @Transactional
    public void verifyAndInsertUser(User user) {
        String email = user.getEmail();
        // 为每个邮箱获取专属锁
        Object lock = lockMap.computeIfAbsent(email, k -> new Object());
        
        synchronized (lock) {
            User existingUser = userRepository.findByEmail(email);
            if (existingUser == null) {
                userRepository.save(user);
            } else {
                throw new UserExistsException("User with email " + email + " already exists.");
            }
        }
        // 可选：移除锁，避免内存泄漏，注意并发安全
        lockMap.remove(email);
    }
}

多实例分布式锁示例（以Redisson为例）： #

@Service
public class UserService {
    @Autowired
    private UserRepository userRepository;
    @Autowired
    private RedissonClient redissonClient;

    @Transactional
    public void verifyAndInsertUser(User user) {
        String email = user.getEmail();
        RLock lock = redissonClient.getLock("user:email:" + email);
        
        try {
            // 尝试获取锁，最多等待5秒，持有锁10秒
            if (lock.tryLock(5, 10, TimeUnit.SECONDS)) {
                User existingUser = userRepository.findByEmail(email);
                if (existingUser == null) {
                    userRepository.save(user);
                } else {
                    throw new UserExistsException("User with email " + email + " already exists.");
                }
            } else {
                throw new RuntimeException("Failed to acquire lock, please try again later.");
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("Lock acquisition interrupted");
        } finally {
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }
}

注意：分布式锁要设置合理的超时时间，避免因服务宕机导致锁无法释放；同时要处理锁获取失败的场景，给用户友好提示。

原代码问题根源 #

默认事务隔离级别（如READ COMMITTED）下，每个线程的事务只能看到其他事务已提交的数据。当线程A执行查询时无对应数据，插入后尚未提交；线程B此时查询也看不到线程A的未提交数据，因此也会执行插入，最终导致重复数据。

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