Spring @Transactional+REPEATABLE_READ实现API月限流的并发一致性问题求解

你遇到的是典型的分布式并发更新场景——多线程/多节点下的计数更新，单纯依赖Spring事务的隔离级别或者JVM同步块确实解决不了，因为同步块只能管单个JVM内的线程，而REPEATABLE_READ隔离级别在"先查后更"的模式下会导致快照读的旧数据被用来更新。下面给你几个可行的解决方案，按实现复杂度和推荐程度排序：

方案1：数据库原子更新（最推荐，无额外依赖） #

核心思路是把计算和更新合并成一条原子SQL语句，避免先查询后更新的竞态条件。数据库的UPDATE语句本身是原子性的，会直接在行上加写锁，确保同一时间只有一个请求能更新这条记录。

伪代码实现（结合Spring） #

@Transactional(propagation = Propagation.REQUIRES_NEW)
public void updateLimitData(String clientId, int currentRecordCount) {
    long startOfMonth = getStartOfMonth();
    long endOfMonth = getEndOfMonth();
    
    // 先尝试更新现有记录
    int updatedRows = jdbcTemplate.update(
        "UPDATE client_monthly_limit SET count = count + ? WHERE client_id = ? AND timestamp_start >= ? AND timestamp_end <= ?",
        currentRecordCount, clientId, startOfMonth, endOfMonth
    );
    
    // 如果没有找到当月记录，插入新记录
    if (updatedRows == 0) {
        jdbcTemplate.update(
            "INSERT INTO client_monthly_limit(client_id, count, timestamp_start, timestamp_end) VALUES (?, ?, ?, ?)",
            clientId, currentRecordCount, startOfMonth, endOfMonth
        );
    }
}

为什么有效？ #

• UPDATE语句是数据库层面的原子操作，执行时会锁定目标行，其他请求必须等待锁释放才能执行。
• 不需要依赖额外组件，完全利用数据库本身的特性，稳定性高。

方案2：乐观锁机制 #

如果你的业务逻辑需要先读取计数做一些额外判断，再更新，可以用乐观锁。在表中新增一个version字段（或者用count本身作为版本，但不推荐），更新时带上版本号，更新失败则重试。

表结构调整 #

新增version列，类型为INT，每次更新自增。

伪代码实现 #

@Transactional(propagation = Propagation.REQUIRES_NEW)
public void updateLimitData(String clientId, int currentRecordCount) {
    long startOfMonth = getStartOfMonth();
    long endOfMonth = getEndOfMonth();
    
    // 读取当前记录和版本号
    ClientMonthlyLimit limit = jdbcTemplate.queryForObject(
        "SELECT * FROM client_monthly_limit WHERE client_id = ? AND timestamp_start >= ? AND timestamp_end <= ?",
        new Object[]{clientId, startOfMonth, endOfMonth},
        new BeanPropertyRowMapper<>(ClientMonthlyLimit.class)
    );
    
    int newCount = limit.getCount() + currentRecordCount;
    int updatedRows = jdbcTemplate.update(
        "UPDATE client_monthly_limit SET count = ?, version = version + 1 WHERE client_id = ? AND timestamp_start >= ? AND timestamp_end <= ? AND version = ?",
        newCount, clientId, startOfMonth, endOfMonth, limit.getVersion()
    );
    
    // 如果更新失败（说明有其他线程先更新了），可以重试或者抛出异常
    if (updatedRows == 0) {
        // 这里可以实现重试逻辑，比如最多重试3次
        throw new ConcurrentUpdateException("Concurrent update detected, please retry");
    }
}

注意点 #

• 需要处理更新失败的情况，通常是重试几次，或者返回给客户端让其重试。
• 适合并发量不是极高的场景，因为高并发下重试次数可能增多。

方案3：分布式锁（适合复杂场景） #

如果你的系统已经引入了Redis等分布式缓存，可以用分布式锁来确保同一客户端当月的更新只有一个线程在执行。比如用Redisson的可重入锁，锁的key可以设为client_limit:{clientId}:{yearMonth}。

伪代码实现（用Redisson） #

@Autowired
private RedissonClient redissonClient;

public void updateLimitData(String clientId, int currentRecordCount) {
    String yearMonth = getYearMonth(); // 比如"202405"
    String lockKey = String.format("client_limit:%s:%s", clientId, yearMonth);
    RLock lock = redissonClient.getLock(lockKey);
    
    try {
        // 尝试获取锁，最多等待5秒，锁持有10秒（根据业务调整）
        if (lock.tryLock(5, 10, TimeUnit.SECONDS)) {
            // 这里用原来的事务逻辑，因为锁已经确保了串行执行
            doUpdateInTransaction(clientId, currentRecordCount);
        } else {
            throw new LockAcquireException("Failed to acquire lock, please retry");
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new RuntimeException("Lock interrupted", e);
    } finally {
        if (lock.isHeldByCurrentThread()) {
            lock.unlock();
        }
    }
}

@Transactional(propagation = Propagation.REQUIRES_NEW)
private void doUpdateInTransaction(String clientId, int currentRecordCount) {
    long startOfMonth = getStartOfMonth();
    long endOfMonth = getEndOfMonth();
    ClientMonthlyLimit limit = fetchFromDB(startOfMonth, endOfMonth, clientId);
    limit.setCount(limit.getCount() + currentRecordCount);
    saveToDB(limit);
}

注意点 #

• 依赖Redis等分布式组件，需要确保组件的高可用性。
• 要处理锁超时、锁释放失败等异常情况，避免死锁。

为什么原来的方案会出问题？ #

你原来的代码用了Isolation.REPEATABLE_READ，在这种隔离级别下，事务中的读取是快照读——第二个线程在第一个线程提交前读取的是事务开始时的快照数据，而不是最新的已提交数据。所以当第一个线程提交更新后，第二个线程还是用旧的快照数据去更新，导致计数被覆盖。

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