分布式系统 并发控制 分布式锁的封装其实大有门道

分布式锁的封装，远比你想象的要复杂

场景引入：

凌晨三点，你被报警电话惊醒——线上订单系统出现了重复扣款，排查日志发现，两个请求同时通过了余额校验，结果各自扣了用户一笔钱，你揉着太阳穴想："明明加了Redis分布式锁啊..."

这就是分布式锁的典型翻车现场，很多人以为锁的封装无非就是lock()和unlock()，但真正高并发的生产环境会教你做人，今天我们就掰开揉碎聊聊这里面的门道。

那些年我们踩过的锁坑

"我的锁怎么自己消失了？"

用Redis的SETNX实现锁时，新手常犯的错误：

# 错误示范
def deduct_balance():
    if redis.setnx("order_lock", 1):  # 获取锁
        try:
            # 业务处理
            time.sleep(10)  # 模拟耗时操作
        finally:
            redis.delete("order_lock")  # 释放锁

问题在于：如果业务处理到一半机器宕机，这个锁就永远无法释放！正确的做法必须设置过期时间：

# 正确姿势
if redis.set("order_lock", 1, nx=True, ex=30):  # 原子性设置值+过期时间
    try:
        process_business()
    finally:
        if redis.get("order_lock") == 1:  # 确保只删除自己的锁
            redis.delete("order_lock")

"锁明明还在，怎么被人抢了？"

更隐蔽的场景是：客户端A获取锁后因GC暂停，锁过期被释放，客户端B获取锁，此时A恢复执行又手动释放了B的锁，解决方案是给锁绑定唯一标识：

// Java示例
String lockId = UUID.randomUUID().toString();
if (redisTemplate.opsForValue().setIfAbsent("lock", lockId, 30, TimeUnit.SECONDS)) {
    try {
        // 业务代码
    } finally {
        // 使用Lua脚本保证原子性检查+删除
        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
        redisTemplate.execute(script, Collections.singletonList("lock"), lockId);
    }
}

进阶封装的核心要点

锁续约机制（Watch Dog）

对于长时间任务，需要动态延长锁有效期，像Redisson的实现逻辑：

1. 获取锁成功后启动后台线程
2. 每隔10秒（默认）检查业务是否完成
3. 未完成则重置锁过期时间为30秒

// Go语言伪代码
func autoRenew() {
    ticker := time.NewTicker(10 * time.Second)
    defer ticker.Stop()
    for {
        select {
        case <-ticker.C:
            if !isBusinessDone {
                redis.Expire("lock", 30*time.Second)
            } else {
                return
            }
        case <-stopChan:
            return
        }
    }
}

可重入锁设计

同一个线程多次获取锁时需要特殊处理：

class RedisLock:
    def __init__(self, client, name):
        self.client = client
        self.name = name
        self._count = 0  # 重入计数器
    def acquire(self):
        if self._count > 0:
            self._count += 1
            return True
        if self.client.set(self.name, threading.get_ident(), nx=True, ex=30):
            self._count = 1
            return True
        return False
    def release(self):
        if self._count <= 0:
            raise RuntimeError("锁未持有")
        self._count -= 1
        if self._count == 0:
            self.client.delete(self.name)

不同场景的锁选型

真实案例：库存超卖解决方案

某电商平台在秒杀活动中采用的分层锁方案：

1. 第一层：本地缓存标记

   // 使用Guava Cache
   LoadingCache<String, Boolean> hotItemCache = CacheBuilder.newBuilder()
       .expireAfterWrite(1, TimeUnit.SECONDS)
       .build(key -> redis.exists("stock_lock_" + key));

2. 第二层：Redis集群锁

   def reduce_stock(item_id):
       if not hotItemCache.get(item_id):
           return False  # 快速失败
       lock_key = f"cluster_lock_{item_id}"
       with redis.lock(lock_key, timeout=500, blocking_timeout=50):
           stock = db.query("SELECT stock FROM items WHERE id=?", item_id)
           if stock > 0:
               db.execute("UPDATE items SET stock=stock-1 WHERE id=?", item_id)

3. 最终兜底：数据库乐观锁

   UPDATE items 
   SET stock = stock - 1 
   WHERE id = ? AND stock = original_stock  -- 通过版本号或CAS机制

写在最后

分布式锁就像分布式系统的安全带，用错了比不用更危险，2025年最新的《分布式系统故障分析报告》显示，35%的分布式事务问题源于锁使用不当，记住三个黄金原则：

1. 原子性：加锁和设置过期时间必须原子操作
2. 可见性：所有节点对锁状态的认知必须一致
3. 容错性：任何单点故障不能导致死锁

下次实现分布式锁时，不妨多问自己：如果此时网络分区怎么办？如果Redis主从切换怎么办？如果客户端进程突然被杀怎么办？想清楚这些问题,你的锁才能真正护住业务的安全线。