如何用 Redis 实现一个分布式锁

场景模拟

一般电子商务网站都会遇到如团购、秒杀、特价之类的活动，而这样的活动有一个共同的特点就是访问量激增、上千甚至上万人抢购一个商品。然而，作为活动商品，库存肯定是很有限的，如何控制库存不让出现超买，以防止造成不必要的损失是众多电子商务网站程序员头疼的问题，这同时也是最基本的问题。

在秒杀系统设计中，

超卖

是一个经典、常见的问题，任何商品都会有数量上限，如何避免成功下订单买到商品的人数不超过商品数量的上限，这是每个抢购活动都要面临的难点。

针对大量的并发请求，我们可以通过 Redis 来抗，也就是说对于库存直接请求 Redis 缓存，不直接请求数据库，如在 Redis 中有 50 个库存，如下：

但不管是缓存还是数据库，在不做任何处理的情况下，都会出现超买的问题，常见的处理方式就是在代码中通过

JVM 加锁

的方式，如下：

server1

@RestController
public class SkillController {

 @Autowired
 private RedisTemplate redisTemplate;

 // 秒杀接口
 @RequestMapping("/deduct_stock")
 public String deductStock() {

 // 加锁
 synchronized (this) {
 int stock = Integer.parseInt(redisTemplate.opsForValue().get("stock").toString());
 if (stock > 0) {
 // 库存 -1
 int realStock = stock - 1;
 // 扣减库存
 redisTemplate.opsForValue().set("stock", realStock + "");
 System.out.println("扣减成功，剩余库存:" + realStock);
 } else {
 System.out.println("扣减失败，库存不足");
 }
 }

 return "8080";
 }
}

当然，在单机情况下确实没有任何问题，但现在绝大多数系统都是分布式系统，就算是 ERP 系统也会部署 2 台机器防止

单点故障

，所以一般情况下一个请求如下：

server2

server2 和 server1 代码基本相同，只是开启了 2 个 JVM 实例。

@RestController
public class SkillController {

 @Autowired
 private RedisTemplate redisTemplate;

 // 秒杀接口
 @RequestMapping("/deduct_stock")
 public String deductStock() {

 // 加锁
 synchronized (this) {
 int stock = Integer.parseInt(redisTemplate.opsForValue().get("stock").toString());
 if (stock > 0) {
 // 库存 -1
 int realStock = stock - 1;
 // 扣减库存
 redisTemplate.opsForValue().set("stock", realStock + "");
 System.out.println("扣减成功，剩余库存:" + realStock);
 } else {
 System.out.println("扣减失败，库存不足");
 }
 }

 return "8090";
 }
}

Nginx

一般来说，前端通过 nginx 请求转发并通过 upstream 实现负载均衡，其关键配置如下：

Jmeter

我们这里通过 Jmeter 来进行并发压测，不会用的参考 

Jmeter 使用

，然后这里提供下载链接：

Jmeter 下载

 （提取码：2hyo）。

并发请求

：1 s 内 200 个请求（模拟高并发），循环 5 次，一共 1000 个总请求。

请求地址

：秒杀的减库存接口。

JVM 锁

了解了上面的配置，然后启动 2 个实例，端口分别为 8080，8090，如下：

如过不知道如何启动 2 个实例的看下面：

注意：要修改启动端口。

使用 JVM 锁也就是同步代码块的方式存在问题，如上面测试的结果如下：

不仅 2 个服务同时存在相同的库存，甚至同一个服务也存在相同的值，很明显在高并发分布式场景下，JVM 层面的锁是不可行的。

Redis SETNX

SETNX

格式：setnx key value

将 key 的值设为 value ，当且仅当 key 不存在。

若给定的 key 已经存在，则 SETNX 不做任何动作。

SETNX 是『SET if Not eXists』(如果不存在，则 SET)的简写。

1、实现一个最简单的分布式锁

@RestController
public class SkillController {

 @Autowired
 private RedisTemplate redisTemplate;

 @RequestMapping("/deduct_stock")
 public String deductStock() {

 // 商品 ID，具体应用中应该是请求传入的
 String lockKey = "lock:product_01";
 // SETNX 加锁
 Boolean result = redisTemplate.opsForValue().setIfAbsent(lockKey, "product");
 
 // 如果为 false 说明这把锁存在，直接返回
 if (!result) {
 // 模拟返回业务
 return "系统繁忙";
 }
 int stock = Integer.parseInt(redisTemplate.opsForValue().get("stock").toString());
 if (stock > 0) {
 // 库存 -1
 int realStock = stock - 1;
 // 扣减库存 模拟其他更多业务操作
 redisTemplate.opsForValue().set("stock", realStock + "");
 System.out.println("扣减成功，剩余库存:" + realStock);
 } else {
 System.out.println("扣减失败，库存不足");
 }

 // 加锁后需要释放锁
 redisTemplate.delete(lockKey);
 
 return "8080";
 }
}

2、存在的问题

① 业务代码异常---死锁

在实际的场景中，一次秒杀过程涉及到很多的业务操作，如果在释放锁之前的某个业务操作抛异常，使得锁没有被释放，那么此时就会存在

死锁

问题。此时这个 key 永远存在于 redis 中，其它线程执行 SETNX 永远失败。

也就是说我们要保证释放锁得到执行，所以要把上面的业务代码放在 

try catch

 或者 

try finally

 中：

② Redis 宕机

但其实上面的代码并不一定能完全解决问题，如果 Redis 宕机或者被重启，同样会导致 finally 中的代码执行失败，结果就同上了。

所以一般我们需要给这个 key 设置过期时间，即：

Boolean result = redisTemplate.opsForValue().setIfAbsent(lockKey, "product");

redisTemplate.expire(lockKey, 10, TimeUnit.SECONDS);

但上面的写法存在原子性问题，所以我们不能分开来写，得合成一条命令：

Boolean result = redisTemplate.opsForValue().setIfAbsent(lockKey, "product", 10, TimeUnit.SECONDS);

③ 高并发可能存在的问题

一般来说，并发量不大的情况下，上面的写法已经满足要求，但对于几千上万的并发，可能会导致接口的响应变慢，比如：

某个请求 A 执行完整个操作需要 15s ，而我们上面设置的超时时间为 10s ，所以此时请求 A 并没有执行完，但由于设置了过期时间把 key 给删掉了，然后这时再来一个请求 B，是可以加锁成功的，且在 8s 内执行完成，而 B 在执行过程中 A 同样也在执行，如果此时 A 可能先于 B 去执行 finally 中的代码把锁给删除了，但 A 删除的锁并不是它的，而是 B 加的锁，同理，当请求 C 加锁后又被请求 B 给释放了，也就是说，这把分布式锁直接无效了（尽管可能性很小），同样会出现超买问题。

这个问题的根本在于：

自己加的锁被被别人释放了

。

因此我们可以确定 value 值唯一性，如 UUID，如下：

但这种方式同样存在原子问题，也就是上图中 ② 处的代码，结果也会导致

自己加的锁被被别人释放

。

Redisson

针对上面的问题，我们可以通过 Redisson 来解决，其使用非常简单，和 JDK 中的 Lock 使用类似，如下：

@RestController
public class RedissonController {

 @Autowired
 private Redisson redisson;

 @Autowired
 private RedisTemplate redisTemplate;

 @RequestMapping("/deduct_stock1")
 public String deductStock() {

 // 商品 ID，具体应用中应该是请求传入的
 String lockKey = "lock:product_01";

 // 获取锁
 RLock lock = redisson.getLock(lockKey);
 // 加锁
 lock.lock();

 try {
 int stock = Integer.parseInt(redisTemplate.opsForValue().get("stock").toString());
 if (stock > 0) {
 // 库存 -1
 int realStock = stock - 1;
 // 扣减库存
 redisTemplate.opsForValue().set("stock", realStock + "");
 System.out.println("扣减成功，剩余库存:" + realStock);
 } else {
 System.out.println("扣减失败，库存不足");
 }
 } finally {
 // 释放锁
 lock.unlock();
 }

 return "8080";
 }
}

再去测试就是正常的了：

Redisson 其原理如下：

Redisson 锁的加锁机制如上图所示，线程去获取锁，获取成功则执行保存数据到 redis 数据库。如果获取失败，则一直通过 while 循环尝试获取锁（可自定义等待时间，超时后返回失败），获取成功后，保存数据到 redis 数据库。

Redisson 提供的分布式锁是支持锁自动续期的（锁续命），也就是说，如果线程仍旧没有执行完，那么 Redisson 会自动给 redis 中的目标 key 延长超时时间，这在 Redisson 中称之为 Watch Dog（看门狗）机制。

那么 

redisson 是怎么实现原子性的

？

当然是 

lua

。不管是加锁操作，还是看门狗机制都是通过 

lua

来保证其原子性。

其加锁调用链路如下：

RedissonLock.lock()--->lockInterruptibly()--->tryAcquire()--->tryLockInnerAsync()

关键代码就在 

tryLockInnerAsync()

 中：

<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
 internalLockLeaseTime = unit.toMillis(leaseTime);
 return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
 // 如果锁不存在，则通过hset设置它的值，并设置过期时间
 &quot;if (redis.call('exists', KEYS[1]) == 0) then &quot; +
 &quot;redis.call('hset', KEYS[1], ARGV[2], 1); &quot; +
 &quot;redis.call('pexpire', KEYS[1], ARGV[1]); &quot; +
 &quot;return nil; &quot; +
 &quot;end; &quot; +
 // 如果锁已存在，其是当前线程，则通过hincrby给数值递增1,即锁的重入
 &quot;if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then &quot; +
 &quot;redis.call('hincrby', KEYS[1], ARGV[2], 1); &quot; +
 &quot;redis.call('pexpire', KEYS[1], ARGV[1]); &quot; +
 &quot;return nil; &quot; +
 &quot;end; &quot; +
 // 如果锁已存在，不是当前线程，则返回过期时间 ttl
 &quot;return redis.call('pttl', KEYS[1]);&quot;,
 Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
}

那么对于锁自动续期如下：

RedissonLock.lock()--->lockInterruptibly()--->tryAcquire()--->scheduleExpirationRenewal()

scheduleExpirationRenewal()

 方法会开启一个子线程去执行自动延期的操作，当然也是执行 

lua

代码，如下，截取关键部分：

// getName()就是当前锁的名字 
RFuture<Boolean> future = commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
 // 判断这个锁 getName() 是否在redis中存在,如果存在就进行 pexpire 延期 默认30s
 &quot;if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then &quot; +
 &quot;redis.call('pexpire', KEYS[1], ARGV[1]); &quot; +
 &quot;return 1; &quot; +
 &quot;end; &quot; +
 &quot;return 0;&quot;,
 Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));


他会判断这个锁 

getName()

 是否在 redis 中存在，如果存在就进行 

pexpire

 延期，默认

lockWatchdogTimeout=30s

，且是每间隔

lockWatchdogTimeout/3=10s

时间，去执行延时操作。

源码：https://gitee.com/javatv/redis.git

参考：

redisson 中的看门狗机制