分布式锁实现原理与最佳实践（6）

05 业务中使用分布式锁的注意点

获取的锁要设置有效期，假设我们未设置key自动过期时间，在Set key value NX 后，如果程序crash或者发生网络分区后无法与Redis节点通信，毫无疑问其他 client 将永远无法获得锁，这将导致死锁，服务出现中断。

SETNX和EXPIRE命令去设置key和过期时间，这也是不正确的，因为你无法保证SETNX和EXPIRE命令的原子性。

自己使用 setnx 实现Redis锁的时候，注意并发情况下不要释放掉别人的锁（业务逻辑执行时间超过锁的过期时间），导致恶性循环。一般：

1）加锁的时候需要指定value的内容是当前进程中的当前线程的唯一标记，不要使用线程ID作为当前线程的锁的标记，因为不同实例上的线程ID可能是一样的。

2）释放锁的逻辑会写在finally ，释放锁时候要判断锁对应的value，而且要使用lua脚本实现原子 del 操作。因为if逻辑判断完之后也可能失效导致删除别人的锁

3）针对扣减库存这个逻辑，lua脚本里面实现Redis比较库存、扣减库存操作的原子性。通过判断Redis Decr命令的返回值即可。此命令会返回扣减后的最新库存，若小于0则表示超卖。

5.1 自己实现分布式锁的坑

✪ setnx不关心锁的顺序导致删除别人的锁

锁失效之后，别人加锁成功，自己把别人的锁删了。

我们无法预估程序执行需要的锁的时间。

public String deductStock() {
    String lockKey = "lock:product_101";
    Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "deltaqin");
    stringRedisTemplate.expire(lockKey, 10, TimeUnit.SECONDS);
    try {
        int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock")); // jedis.get("stock")
        if (stock > 0) {
            int realStock = stock - 1;
            stringRedisTemplate.opsForValue().set("stock", realStock + ""); // jedis.set(key,value)
            System.out.println("扣减成功，剩余库存:" + realStock);
        } else {
            System.out.println("扣减失败，库存不足");
        }
    } finally {
        stringRedisTemplate.delete(lockKey);
    }
    return "end";
}

✪ setnx关心锁的顺序还是删除了别人的锁

并发会卡在各种地方，卡住的时候过期了，就会删掉别人加的锁：

错误的原因还是因为解锁的逻辑不是原子性的，这里可以参考Redisson的解锁逻辑使用lua脚本实现。

public String deductStock() {
    String lockKey = "lock:product_101";
    String clientId = UUID.randomUUID().toString();
    Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, clientId, 30, TimeUnit.SECONDS); //jedis.setnx(k,v)
    if (!result) {
        return "error_code";
    }
    try {
        int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock")); // jedis.get("stock")
        if (stock > 0) {
            int realStock = stock - 1;
            stringRedisTemplate.opsForValue().set("stock", realStock + ""); // jedis.set(key,value)
            System.out.println("扣减成功，剩余库存:" + realStock);
        } else {
            System.out.println("扣减失败，库存不足");
        }
    } finally {
        if (clientId.equals(stringRedisTemplate.opsForValue().get(lockKey))) {
            // 卡在这里，锁过期了，其他线程又可以加锁，此时又把其他线程新加的锁删掉了
            stringRedisTemplate.delete(lockKey);
        }
    }
    return "end";
}

⍟ 解决办法

这种问题解决的办法就是使用锁续命，比如使用一个定时任务间隔小于锁的超时时间，每隔一段时间就给锁续命，除非线程自己主动删除。这也是Redisson的实现思路。

5.2 锁优化：分段加锁逻辑

针对一个商品，要开启秒杀的时候，会将商品的库存预先加载到Redis缓存中，比如有100个库存，此时可以分为5个key，每一个key有20个库存。可以把分布式锁的性能提升5倍。

例如：

• product_10111_stock = 100
  

• product_10111_stock1 = 20
  
• product_10111_stock2 = 20
  
• product_10111_stock3 = 20
  
• product_10111_stock4 = 20
  
• product_10111_stock5 = 20
  

请求来了可以随机可以轮询，扣减完之后就标记不要下次再分配到这个库存。

06 分布式锁的真相与选择

6.1 分布式锁的真相

需要满足的几个特性

• 互斥：不同线程、进程互斥。
  
• 超时机制：临界区代码耗时导致，网络原因导致。可以使用额外的线程续命保证。
  
• 完备的锁接口：阻塞的和非阻塞的接口都要有，lock和tryLock。
  
• 可重入性：当前请求的节点+ 线程唯一标识。
  
• 公平性：锁唤醒时候，按照顺序唤醒。
  
• 正确性：进程内的锁不会因为报错死锁，因为崩溃的时候整个进程都会结束。但是多实例部署时死锁就很容易发生，如果粗暴使用超时机制解决死锁问题，就默认了下面这个假设：
  

• 锁的超时时间 >> 获取锁的时延 + 执行临界区代码的时间 + 各种进程的暂停（比如 GC）
  
• 但上述假设其实无法保证的。
  

将分布式锁定位为，可以容忍非常小概率互斥语义失效场景下的锁服务。一般来说，一个分布式锁服务，它的正确性要求越高，性能可能就会越低。

6.2 分布式锁的选择

• 数据库：db操作性能较差，并且有锁表的风险，一般不考虑。
  

• 优点：实现简单、易于理解
  
• 缺点：对数据库压力大
  

• Redis：适用于并发量很大、性能要求很高而可靠性问题可以通过其他方案去弥补的场景。
  

• 优点：易于理解
  
• 缺点：自己实现、不支持阻塞
  
• Redisson：相对于Jedis其实更多用在分布式的场景。
  

• 优点：提供锁的方法，可阻塞
  

• Zookeeper：适用于高可靠（高可用），而并发量不是太高的场景。
  

• 优点：支持阻塞
  
• 缺点：需理解Zookeeper、程序复杂
  

• Curator
  

• 优点：提供锁的方法
  
• 缺点：Zookeeper，强一致，慢
  

• Etcd：安全和可靠性上有保证，但是比较重。
  

不推荐自己编写的分布式锁，推荐使用Redisson和Curator实现的分布式锁。