分布式锁用Redis还是Zookeeper？

　　为什么用分布式锁？在讨论这个问题之前，我们先来看一个业务场景。
　　图片来自 Pexels
　　为什么用分布式锁？
　　系统 A 是一个电商系统，目前是一台机器部署，系统中有一个用户下订单的接口，但是用户下订单之前一定要去检查一下库存，确保库存足够了才会给用户下单。
　　由于系统有一定的并发，所以会预先将商品的库存保存在 Redis 中，用户下单的时候会更新 Redis 的库存。
　　此时系统架构如下：
　　但是这样一来会产生一个问题：假如某个时刻，Redis 里面的某个商品库存为 1。
　　此时两个请求同时到来，其中一个请求执行到上图的第 3 步，更新数据库的库存为 0，但是第 4 步还没有执行。
　　而另外一个请求执行到了第 2 步，发现库存还是 1，就继续执行第 3 步。这样的结果，是导致卖出了 2 个商品，然而其实库存只有 1 个。
　　很明显不对啊！这就是典型的库存超卖问题。此时，我们很容易想到解决方案：用锁把 2、3、4 步锁住，让他们执行完之后，另一个线程才能进来执行第 2 步。
　　按照上面的图，在执行第 2 步时，使用 Java 提供的 Synchronized 或者 ReentrantLock 来锁住，然后在第 4 步执行完之后才释放锁。
　　这样一来，2、3、4 这 3 个步骤就被＂锁＂住了，多个线程之间只能串行化执行。
　　但是好景不长，整个系统的并发飙升，一台机器扛不住了。现在要增加一台机器，如下图：
　　增加机器之后，系统变成上图所示，我的天！假设此时两个用户的请求同时到来，但是落在了不同的机器上，那么这两个请求是可以同时执行了，还是会出现库存超卖的问题。
　　为什么呢？因为上图中的两个 A 系统，运行在两个不同的 JVM 里面，他们加的锁只对属于自己 JVM 里面的线程有效，对于其他 JVM 的线程是无效的。
　　因此，这里的问题是：Java 提供的原生锁机制在多机部署场景下失效了，这是因为两台机器加的锁不是同一个锁（两个锁在不同的 JVM 里面）。
　　那么，我们只要保证两台机器加的锁是同一个锁，问题不就解决了吗？此时，就该分布式锁隆重登场了。
　　分布式锁的思路是：在整个系统提供一个全局、唯一的获取锁的＂东西＂，然后每个系统在需要加锁时，都去问这个＂东西＂拿到一把锁，这样不同的系统拿到的就可以认为是同一把锁。
　　至于这个＂东西＂，可以是 Redis、Zookeeper，也可以是数据库。文字描述不太直观，我们来看下图：
　　通过上面的分析，我们知道了库存超卖场景在分布式部署系统的情况下使用 Java 原生的锁机制无法保证线程安全，所以我们需要用到分布式锁的方案。
　　那么，如何实现分布式锁呢？接着往下看！
　　这里整理了关于分布式锁的视频讲解，保姆级视频，可以订阅免费的公开课，每晚8点直播讲解，免费订阅即可学习：C/C++Linux服务器开发/后台架构师【零声教育】-学习视频教程-腾讯课堂
　　基于 Redis 实现分布式锁
　　上面分析为啥要使用分布式锁了，这里我们来具体看看分布式锁落地的时候应该怎么样处理。
　　①常见的一种方案就是使用 Redis 做分布式锁
　　使用 Redis 做分布式锁的思路大概是这样的：在 Redis 中设置一个值表示加了锁，然后释放锁的时候就把这个 Key 删除。
　　具体代码是这样的：// 获取锁  // NX是指如果key不存在就成功，key存在返回false，PX可以指定过期时间  SET anyLock unique_value NX PX 30000      // 释放锁：通过执行一段lua脚本  // 释放锁涉及到两条指令，这两条指令不是原子性的  // 需要用到redis的lua脚本支持特性，redis执行lua脚本是原子性的  if redis.call(＂get＂,KEYS[1]) == ARGV[1] then  return redis.call(＂del＂,KEYS[1])  else  return 0  end
　　这种方式有几大要点：一定要用 SET key value NX PX milliseconds 命令。如果不用，先设置了值，再设置过期时间，这个不是原子性操作，有可能在设置过期时间之前宕机，会造成死锁(Key 永久存在)Value 要具有唯一性。这个是为了在解锁的时候，需要验证 Value 是和加锁的一致才删除 Key。
　　这时避免了一种情况：假设 A 获取了锁，过期时间 30s，此时 35s 之后，锁已经自动释放了，A 去释放锁，但是此时可能 B 获取了锁。A 客户端就不能删除 B 的锁了。
　　除了要考虑客户端要怎么实现分布式锁之外，还需要考虑 Redis 的部署问题。
　　Redis 有 3 种部署方式：单机模式Master-Slave+Sentinel 选举模式Redis Cluster 模式
　　使用 Redis 做分布式锁的缺点在于：如果采用单机部署模式，会存在单点问题，只要 Redis 故障了。加锁就不行了。
　　采用 Master-Slave 模式，加锁的时候只对一个节点加锁，即便通过 Sentinel 做了高可用，但是如果 Master 节点故障了，发生主从切换，此时就会有可能出现锁丢失的问题。
　　基于以上的考虑，Redis 的作者也考虑到这个问题，他提出了一个 RedLock 的算法。
　　这个算法的意思大概是这样的：假设 Redis 的部署模式是 Redis Cluster，总共有 5 个 Master 节点。
　　通过以下步骤获取一把锁：获取当前时间戳，单位是毫秒。轮流尝试在每个 Master 节点上创建锁，过期时间设置较短，一般就几十毫秒。尝试在大多数节点上建立一个锁，比如 5 个节点就要求是 3 个节点（n / 2 +1）。客户端计算建立好锁的时间，如果建立锁的时间小于超时时间，就算建立成功了。要是锁建立失败了，那么就依次删除这个锁。只要别人建立了一把分布式锁，你就得不断轮询去尝试获取锁。
　　但是这样的这种算法还是颇具争议的，可能还会存在不少的问题，无法保证加锁的过程一定正确。
　　②另一种方式：Redisson
　　此外，实现 Redis 的分布式锁，除了自己基于 Redis Client 原生 API 来实现之外，还可以使用开源框架：Redission。
　　Redisson 是一个企业级的开源 Redis Client，也提供了分布式锁的支持。我也非常推荐大家使用，为什么呢？
　　回想一下上面说的，如果自己写代码来通过 Redis 设置一个值，是通过下面这个命令设置的：SET anyLock unique_value NX PX 30000
　　这里设置的超时时间是 30s，假如我超过 30s 都还没有完成业务逻辑的情况下，Key 会过期，其他线程有可能会获取到锁。
　　这样一来的话，第一个线程还没执行完业务逻辑，第二个线程进来了也会出现线程安全问题。
　　所以我们还需要额外的去维护这个过期时间，太麻烦了~我们来看看 Redisson 是怎么实现的？
　　先感受一下使用 Redission 的爽：Config config = new Config();  config.useClusterServers()  .addNodeAddress(＂redis://192.168.31.101:7001＂)  .addNodeAddress(＂redis://192.168.31.101:7002＂)  .addNodeAddress(＂redis://192.168.31.101:7003＂)  .addNodeAddress(＂redis://192.168.31.102:7001＂)  .addNodeAddress(＂redis://192.168.31.102:7002＂)  .addNodeAddress(＂redis://192.168.31.102:7003＂);    RedissonClient redisson = Redisson.create(config);      RLock lock = redisson.getLock(＂anyLock＂);  lock.lock();  lock.unlock();
　　就是这么简单，我们只需要通过它的 API 中的 Lock 和 Unlock 即可完成分布式锁，他帮我们考虑了很多细节：Redisson 所有指令都通过 Lua 脚本执行，Redis 支持 Lua 脚本原子性执行。Redisson 设置一个 Key 的默认过期时间为 30s，如果某个客户端持有一个锁超过了 30s 怎么办？Redisson 中有一个 Watchdog 的概念，翻译过来就是看门狗，它会在你获取锁之后，每隔 10s 帮你把 Key 的超时时间设为 30s。
　　这样的话，就算一直持有锁也不会出现 Key 过期了，其他线程获取到锁的问题了。Redisson 的＂看门狗＂逻辑保证了没有死锁发生。(如果机器宕机了，看门狗也就没了。此时就不会延长 Key 的过期时间，到了 30s 之后就会自动过期了，其他线程可以获取到锁)
　　这里稍微贴出来其实现代码：// 加锁逻辑  private  RFuture tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) {      if (leaseTime != -1) {   return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);      }      // 调用一段lua脚本，设置一些key、过期时间      RFuture ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);      ttlRemainingFuture.addListener(new FutureListener() {          @Override   public void operationComplete(Future future) throws Exception {              if (!future.isSuccess()) {   return;              }                Long ttlRemaining = future.getNow();              // lock acquired              if (ttlRemaining == null) {                  // 看门狗逻辑                  scheduleExpirationRenewal(threadId);              }          }      });   return ttlRemainingFuture;  }       RFuture tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand command) {      internalLockLeaseTime = unit.toMillis(leaseTime);     return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,   ＂if (redis.call(＂exists＂, KEYS[1]) == 0) then ＂ +   ＂redis.call(＂hset＂, KEYS[1], ARGV[2], 1); ＂ +   ＂redis.call(＂pexpire＂, KEYS[1], ARGV[1]); ＂ +   ＂return nil; ＂ +   ＂end; ＂ +   ＂if (redis.call(＂hexists＂, KEYS[1], ARGV[2]) == 1) then ＂ +   ＂redis.call(＂hincrby＂, KEYS[1], ARGV[2], 1); ＂ +   ＂redis.call(＂pexpire＂, KEYS[1], ARGV[1]); ＂ +   ＂return nil; ＂ +   ＂end; ＂ +   ＂return redis.call(＂pttl＂, KEYS[1]);＂,                  Collections.