从一次秒杀超卖问题聊一聊分布式锁
前段时间，我的一个朋友小 A 找我说，前几天他们公司出现了一个比较严重的线上问题，原因是春节系统进行促销，很多商品参与秒杀活动，供应商原先预定的库存，结果由于系统问题导致了超卖，结果被老板劈头盖脸的一顿骂，这是一个很典型的高并发下线程安全导致程序异常的问题，下面通过一段代码来复现。
代码复现
在我们日常的开发中，如果平常系统的 QPS 比较低，不考虑高并发的情况下，假如我们需要做一个减库存的操作，一般我们的代码可以如下实现，为了方便起见，我将库存信息存放到 redis 中。
那么对应的代码如下：
@Slf4j
@RequestMapping("/sku")
@RestController
public class SkuController {
@Autowired
RedisTemplate redisTemplate;
@GetMapping("restock")
public String subStock(String id) {
int stock = Integer.parseInt(String.valueOf(redisTemplate.opsForValue().get("stock:" + id)));
if (stock > 0) {
//减库存
int leaveStock = stock - 1;
redisTemplate.opsForValue().set("stock:" + id, leaveStock);
log.info("扣除库存成功！剩余库存:{}", leaveStock);
// 其他业务操作...
} else {
log.error("扣除库存失败！库存不足");
}
return "ok";
}
}可以分析下，上面的代码，乍看好像没啥问题，不过如果是高并发的情况下，就可能会有问题了，你会发现库存的各种错乱，最后无奈要砸键盘了。
分布式锁
在某些场景中，多个进程必须以互斥的方式独占共享资源，这时用分布式锁是最直接有效的。
随着技术快速发展，数据规模增大，分布式系统越来越普及，一个应用往往会部署在多台机器上（多节点），在有些场景中，为了保证数据不重复，要求在同一时刻，同一任务只在一个节点上运行，即保证某一方法同一时刻只能被一个线程执行。在单机环境中，应用是在同一进程下的，只需要保证单进程多线程环境中的线程安全性，通过 JAVA 提供的 volatile、ReentrantLock、synchronized 以及 concurrent 并发包下一些线程安全的类等就可以做到。而在多机部署环境中，不同机器不同进程，就需要在多进程下保证线程的安全性了。因此，分布式锁应运而生。
以往的工作中看到或用到几种实现方案，有基于 zk 的，也有基于 redis 的。由于实现上逻辑不严谨，线上时不时会爆出几个死锁 case。那么，究竟什么样的分布式锁实现，才算是比较好的方案？
常见分布式锁方案对比
| 分类 | 方案 | 实现原理 | 优点 | 缺点 |
| 基于数据库 | 基于 mysql 表唯一索引 | 1.表增加唯一索引 2.加锁：执行 insert 语句，若报错，则表明加锁失败 3.解锁：执行 delete 语句 | 完全利用 DB 现有能力，实现简单 | 1.锁无超时自动失效机制，有死锁风险 2.不支持锁重入，不支持阻塞等待 3.操作数据库开销大，性能不高 |
| 基于数据库 | 基于 MongoDB findAndModify 原子操作 | 1.加锁：执行 findAndModify 原子命令查找 document，若不存在则新增 2.解锁：删除 document | 实现也很容易，较基于 MySQL 唯一索引的方案，性能要好很多 | 1.大部分公司数据库用 MySQL，可能缺乏相应的 MongoDB 运维、开发人员 2.锁无超时自动失效机制 |
| 基于分布式协调系统 | 基于 ZooKeeper | 1.加锁：在/lock 目录下创建临时有序节点，判断创建的节点序号是否最小。若是，则表示获取到锁；否，则则 watch /lock 目录下序号比自身小的前一个节点 2.解锁：删除节点 | 1.由 zk 保障系统高可用 2.Curator 框架已原生支持系列分布式锁命令，使用简单 | 需单独维护一套 zk 集群，维保成本高 |
| 基于缓存 | 基于 redis 命令 | 1. 加锁：执行 setnx，若成功再执行 expire 添加过期时间 2. 解锁：执行 delete 命令 | 实现简单，相比数据库和分布式系统的实现，该方案最轻，性能最好 | 1.setnx 和 expire 分 2 步执行，非原子操作；若 setnx 执行成功，但 expire 执行失败，就可能出现死锁 2.delete 命令存在误删除非当前线程持有的锁的可能 3.不支持阻塞等待、不可重入 |
| 基于缓存 | 基于 redis Lua 脚本能力 | 1. 加锁：执行 SET lock_name random_value EX seconds NX 命令 2. 解锁：执行 Lua 脚本，释放锁时验证 random_value -- ARGV[1]为 random_value, KEYS[1]为 lock_name if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("del",KEYS[1]) else return 0 end | 同上；实现逻辑上也更严谨，除了单点问题，生产环境采用用这种方案，问题也不大。 | 不支持锁重入，不支持阻塞等待 |
| 基于缓存 | redisson 框架 | 1. 加锁：执行 lock()方法 2. 解锁：执行 unlock()方法 | 1.逻辑严谨，支持各种集群架构，api 操作简单。 2. 底层依旧使用了 lua 脚本，支持锁重入。 3.支持各种锁优化场景，如读写锁，红锁等 | 需要单独引入 redisson 依赖 |
表格中对比了几种常见的方案，redis+lua 基本可应付工作中分布式锁的需求。然而，线上一般推荐使用 redisson 框架，redisson 保持了简单易用、支持锁重入、支持阻塞等待、Lua 脚本原子操作，作者在实现的时候也非常的巧妙，而且考虑到非常多的复杂业务场景。
分布式锁需满足四个条件
首先，为了确保分布式锁可用，我们至少要确保锁的实现同时满足以下四个条件：
1.互斥性。在任意时刻，只有一个客户端能持有锁。2.不会发生死锁。即使有一个客户端在持有锁的期间崩溃而没有主动解锁，也能保证后续其他客户端能加锁。3.解铃还须系铃人。加锁和解锁必须是同一个客户端，客户端自己不能把别人加的锁给解了，即不能误解锁。4.具有容错性。只要大多数 Redis 节点正常运行，客户端就能够获取和释放锁。
Redisson 分布式锁的实现
Redisson 分布式重入锁用法
Redisson 支持单点模式、主从模式、哨兵模式、集群模式，这里以单点模式为例：
// 1.构造redisson实现分布式锁必要的Config
Config config = new Config();
config.useSingleServer().setAddress("redis://127.0.0.1:5379").setPassword("123456").setDatabase(0);
// 2.构造RedissonClient
RedissonClient redissonClient = Redisson.create(config);
// 3.获取锁对象实例（无法保证是按线程的顺序获取到）
RLock rLock = redissonClient.getLock(lockKey);
try {
/**
* 4.尝试获取锁
* waitTimeout 尝试获取锁的最大等待时间，超过这个值，则认为获取锁失败
* leaseTime 锁的持有时间,超过这个时间锁会自动失效（值应设置为大于业务处理的时间，确保在锁有效期内业务能处理完）
*/
boolean res = rLock.tryLock((long)waitTimeout, (long)leaseTime, TimeUnit.SECONDS);
if (res) {
//成功获得锁，在这里处理业务
}
} catch (Exception e) {
throw new RuntimeException("aquire lock fail");
}finally{
//无论如何, 最后都要解锁
rLock.unlock();
}redisson 这个框架重度依赖了 Lua 脚本和 Netty，代码很牛逼，各种 Future 及 FutureListener 的异步、同步操作转换。
自己先思考下,如果要手写一个分布式锁组件，怎么做？肯定要定义 2 个接口：加锁、解锁；大道至简，redisson 的作者就是在加锁和解锁的执行层面采用 Lua 脚本，逼格高，而且重要有原子性保证啊。当然，redisson 的作者毕竟牛逼，加锁和解锁过程中还巧妙地利用了 redis 的发布订阅功能，后面会讲到。下面先对加锁和解锁 Lua 脚本了解下。
1、加锁&解锁 Lua 脚本
| 参数 | 示例值 | 含义 |
| KEY 个数 | 1 | KEY 个数 |
| KEYS[1] | my_first_lock_name | 锁名 |
| ARGV[1] | 60000 | 持有锁的有效时间：毫秒 |
| ARGV[2] | 58c62432-bb74-4d14-8a00-9908cc8b828f:1 | 唯一标识：获取锁时 set 的唯一值，实现上为 redisson 客户端 ID(UUID)+线程 ID |
•脚本内容
-- 若锁不存在：则新增锁，并设置锁重入计数为1、设置锁过期时间
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;
-- 若锁存在，且唯一标识也匹配：则表明当前加锁请求为锁重入请求，故锁重入计数+1，并再次设置锁过期时间
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]);•实现原理
Q：返回 nil、返回剩余过期时间有什么目的？ A：当且仅当返回 nil，才表示加锁成功；客户端需要感知加锁是否成功的结果
2、解锁 Lua 脚本
•脚本入参
| 参数 | 示例值 | 含义 |
| KEY 个数 | 2 | KEY 个数 |
| KEYS[1] | my_first_lock_name | 锁名 |
| KEYS[2] | redisson_lock__channel:{my_first_lock_name} | 解锁消息 PubSub 频道 |
| ARGV[1] | 0 | redisson 定义 0 表示解锁消息 |
| ARGV[2] | 30000 | 设置锁的过期时间；默认值 30 秒 |
| ARGV[3] | 58c62432-bb74-4d14-8a00-9908cc8b828f:1 | 唯一标识；同加锁流程 |
•脚本内容
-- 若锁不存在：则直接广播解锁消息，并返回1
if (redis.call('exists', KEYS[1]) == 0) then
redis.call('publish', KEYS[2], ARGV[1]);
return 1;
end;
-- 若锁存在，但唯一标识不匹配：则表明锁被其他线程占用，当前线程不允许解锁其他线程持有的锁
if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then
return nil;
end;
-- 若锁存在，且唯一标识匹配：则先将锁重入计数减1
local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1);
if (counter > 0) then
-- 锁重入计数减1后还大于0：表明当前线程持有的锁还有重入，不能进行锁删除操作，但可以友好地帮忙设置下过期时期
redis.call('pexpire', KEYS[1], ARGV[2]);
return 0;
else
-- 锁重入计数已为0：间接表明锁已释放了。直接删除掉锁，并广播解锁消息，去唤醒那些争抢过锁但还处于阻塞中的线程
redis.call('del', KEYS[1]);
redis.call('publish', KEYS[2], ARGV[1]);
return 1;
end;
return nil;•主要原理
Q1：广播解锁消息有什么用？ A：是为了通知其他争抢锁阻塞住的线程，从阻塞中解除，并再次去争抢锁。
Q2：返回值 0、1、nil 有什么不一样？ A：当且仅当返回 1，才表示当前请求真正触发了解锁 Lua 脚本；但客户端又并不关心解锁请求的返回值，好像没什么用？
分析源码
1、加锁流程源码
读加锁源码时，可以把 tryAcquire(leaseTime, unit, threadId)方法直接视为执行加锁 Lua 脚本。直接进入 org.redisson.RedissonLock#tryLock(long, long, java.util.concurrent.TimeUnit)源码
@Override
public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
// 获取锁能容忍的最大等待时长
long time = unit.toMillis(waitTime);
long current = System.currentTimeMillis();
final long threadId = Thread.currentThread().getId();
// 【核心点1】尝试获取锁，若返回值为null，则表示已获取到锁
Long ttl = tryAcquire(leaseTime, unit, threadId);
// lock acquired
if (ttl == null) {
return true;
}
// 还可以容忍的等待时长=获取锁能容忍的最大等待时长 - 执行完上述操作流逝的时间
time -= (System.currentTimeMillis() - current);
if (time <= 0) {
acquireFailed(threadId);
return false;
}
current = System.currentTimeMillis();
// 【核心点2】订阅解锁消息，见org.redisson.pubsub.LockPubSub#onMessage
/**
* 4.订阅锁释放事件，并通过await方法阻塞等待锁释放，有效的解决了无效的锁申请浪费资源的问题：
* 基于信息量，当锁被其它资源占用时，当前线程通过 Redis 的 channel 订阅锁的释放事件，一旦锁释放会发消息通知待等待的线程进行竞争
* 当 this.await返回false，说明等待时间已经超出获取锁最大等待时间，取消订阅并返回获取锁失败
* 当 this.await返回true，进入循环尝试获取锁
*/
final RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
//await 方法内部是用CountDownLatch来实现阻塞，获取subscribe异步执行的结果（应用了Netty 的 Future）
if (!await(subscribeFuture, time, TimeUnit.MILLISECONDS)) {
if (!subscribeFuture.cancel(false)) {
subscribeFuture.addListener(new FutureListener<RedissonLockEntry>() {
@Override
public void operationComplete(Future<RedissonLockEntry> future) throws Exception {
if (subscribeFuture.isSuccess()) {
unsubscribe(subscribeFuture, threadId);
}
}
});
}
acquireFailed(threadId);
return false;
}
// 订阅成功
try {
// 还可以容忍的等待时长=获取锁能容忍的最大等待时长 - 执行完上述操作流逝的时间
time -= (System.currentTimeMillis() - current);
if (time <= 0) {
// 超出可容忍的等待时长，直接返回获取锁失败
acquireFailed(threadId);
return false;
}
while (true) {
long currentTime = System.currentTimeMillis();
// 尝试获取锁；如果锁被其他线程占用，就返回锁剩余过期时间【同上】
ttl = tryAcquire(leaseTime, unit, threadId);
// lock acquired
if (ttl == null) {
return true;
}
time -= (System.currentTimeMillis() - currentTime);
if (time <= 0) {
acquireFailed(threadId);
return false;
}
// waiting for message
currentTime = System.currentTimeMillis();
// 【核心点3】根据锁TTL，调整阻塞等待时长；
// 注意：这里实现非常巧妙，1、latch其实是个信号量Semaphore，调用其tryAcquire方法会让当前线程阻塞一段时间，避免了在while循环中频繁请求获取锁；
//2、该Semaphore的release方法，会在订阅解锁消息的监听器消息处理方法org.redisson.pubsub.LockPubSub#onMessage调用；当其他线程释放了占用的锁，会广播解锁消息，监听器接收解锁消息，并释放信号量，最终会唤醒阻塞在这里的线程。
if (ttl >= 0 && ttl < time) {
getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
} else {
getEntry(threadId).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
}
time -= (System.currentTimeMillis() - currentTime);
if (time <= 0) {
acquireFailed(threadId);
return false;
}
}
} finally {
// 取消解锁消息的订阅
unsubscribe(subscribeFuture, threadId);
}
}接下的再获取锁方法 tryAcquire 的实现，真的就是执行 Lua 脚本！
private Long tryAcquire(long leaseTime, TimeUnit unit, long threadId) {
// tryAcquireAsync异步执行Lua脚本，get方法同步获取返回结果
return get(tryAcquireAsync(leaseTime, unit, threadId));
}
// 见org.redisson.RedissonLock#tryAcquireAsync
private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) {
if (leaseTime != -1) {
// 实质是异步执行加锁Lua脚本
return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
}
RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
ttlRemainingFuture.addListener(new FutureListener<Long>() {
@Override
public void operationComplete(Future<Long> future) throws Exception {
//先判断这个异步操作有没有执行成功，如果没有成功，直接返回，如果执行成功了，就会同步获取结果
if (!future.isSuccess()) {
return;
}
Long ttlRemaining = future.getNow();
// lock acquired
//如果ttlRemaining为null，则会执行一个定时调度的方法scheduleExpirationRenewal
if (ttlRemaining == null) {
scheduleExpirationRenewal(threadId);
}
}
});
return ttlRemainingFuture;
}
// 见org.redisson.RedissonLock#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,
"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.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
}加锁过程小结
1、锁其实也是一种资源，各线程争抢锁操作对应到 redisson 中就是争抢着去创建一个 hash 结构，谁先创建就代表谁获得锁；hash 的名称为锁名，hash 里面内容仅包含一条键值对，键为 redisson 客户端唯一标识+持有锁线程 id，值为锁重入计数；给 hash 设置的过期时间就是锁的过期时间。放个图直观感受下：
2、加锁流程核心就 3 步
Step1：尝试获取锁，这一步是通过执行加锁 Lua 脚本来做；
Step2：若第一步未获取到锁，则去订阅解锁消息，当获取锁到剩余过期时间后，调用信号量方法阻塞住，直到被唤醒或等待超时
Step3：一旦持有锁的线程释放了锁，就会广播解锁消息。于是，第二步中的解锁消息的监听器会释放信号量，获取锁被阻塞的那些线程就会被唤醒，并重新尝试获取锁。
比如 RedissonLock 中的变量 internalLockLeaseTime,默认值是 30000 毫秒，还有调用 tryLockInnerAsync()传入的一个从连接管理器获取的 getLockWatchdogTimeout(),他的默认值也是 30000 毫秒，这些都和 redisson 官方文档所说的 watchdog 机制有关，看门狗，还是很形象的描述这一机制，那么看门狗到底做了什么，为什么这么做,来看下核心代码.
先思考一个问题，假设在一个分布式环境下，多个服务实例请求获取锁，其中服务实例 1 成功获取到了锁，在执行业务逻辑的过程中，服务实例突然挂掉了或者 hang 住了，那么这个锁会不会释放，什么时候释放？回答这个问题，自然想起来之前我们分析的 lua 脚本，其中第一次加锁的时候使用 pexpire 给锁 key 设置了过期时间，默认 30000 毫秒，由此来看如果服务实例宕机了，锁最终也会释放，其他服务实例也是可以继续获取到锁执行业务。但是要是 30000 毫秒之后呢，要是服务实例 1 没有宕机但是业务执行还没有结束，所释放掉了就会导致线程问题，这个 redisson 是怎么解决的呢？这个就一定要实现自动延长锁有效期的机制。
异步执行完 lua 脚本执行完成之后，设置了一个监听器，来处理异步执行结束之后的一些工作。在操作完成之后会去执行 operationComplete 方法，先判断这个异步操作有没有执行成功，如果没有成功，直接返回，如果执行成功了，就会同步获取结果，如果 ttlRemaining 为 null，则会执行一个定时调度的方法 scheduleExpirationRenewal,回想一下之前的 lua 脚本，当加锁逻辑 处理结束，返回了一个 nil;如此说来 就一定会走定时任务了。来看下定时调度 scheduleExpirationRenewal 代码。
private void scheduleExpirationRenewal(final long threadId) {
if (expirationRenewalMap.containsKey(getEntryName())) {
return;
}
Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
@Override
public void run(Timeout timeout) throws Exception {
RFuture<Boolean> future = commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return 1; " +
"end; " +
"return 0;",
Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
future.addListener(new FutureListener<Boolean>() {
@Override
public void operationComplete(Future<Boolean> future) throws Exception {
expirationRenewalMap.remove(getEntryName());
if (!future.isSuccess()) {
log.error("Can't update lock " + getName() + " expiration", future.cause());
return;
}
if (future.getNow()) {
// reschedule itself
scheduleExpirationRenewal(threadId);
}
}
});
}
}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
if (expirationRenewalMap.putIfAbsent(getEntryName(), task) != null) {
task.cancel();
}
}首先，会先判断在 expirationRenewalMap 中是否存在了 entryName，这是个 map 结构，主要还是判断在这个服务实例中的加锁客户端的锁 key 是否存在，如果已经存在了，就直接返回；第一次加锁，肯定是不存在的，接下来就是搞了一个 TimeTask，延迟 internalLockLeaseTime/3 之后执行，这里就用到了文章一开始就提到奇妙的变量，算下来就是大约 10 秒钟执行一次，调用了一个异步执行的方法
如图也是调用异步执行了一段 lua 脚本，首先判断这个锁 key 的 map 结构中是否存在对应的 key8a9649f5-f5b5-48b4-beaa-d0c24855f9ab:anyLock:1,如果存在，就直接调用 pexpire 命令设置锁 key 的过期时间,默认 30000 毫秒。
OK，现在思路就清晰了，在上面任务调度的方法中，也是异步执行并且设置了一个监听器，在操作执行成功之后，会回调这个方法，如果调用失败会打一个错误日志并返回，更新锁过期时间失败；然后获取异步执行的结果，如果为 true，就会调用本身，如此说来又会延迟 10 秒钟去执行这段逻辑，所以，这段逻辑在你成功获取到锁之后，会每隔十秒钟去执行一次，并且，在锁 key 还没有失效的情况下，会把锁的过期时间继续延长到 30000 毫秒，也就是说只要这台服务实例没有挂掉，并且没有主动释放锁，看门狗都会每隔十秒给你续约一下，保证锁一直在你手中。完美的操作。
到现在来说，加锁，锁自动延长过期时间，都 OK 了，然后就是说在你执行业务，持有锁的这段时间，别的服务实例来尝试加锁又会发生什么情况呢？或者当前客户端的别的线程来获取锁呢？很显然，肯定会阻塞住，我们来通过代码看看是怎么做到的。还是把眼光放到之前分析的那段加锁 lua 代码上，当加锁的锁 key 存在的时候并且锁 key 对应的 map 结构中当前客户端的唯一 key 也存在时，会去调用 hincrby 命令，将唯一 key 的值自增一，并且会 pexpire 设置 key 的过期时间为 30000 毫秒，然后返回 nil,可以想象这里也是加锁成功的，也会继续去执行定时调度任务，完成锁 key 过期时间的续约，这里呢，就实现了锁的可重入性。
那么当以上这种情况也没有发生呢，这里就会直接返回当前锁的剩余有效期，相应的也不会去执行续约逻辑。此时一直返回到上面的方法，如果加锁成功就直接返回;否则就会进入一个死循环，去尝试加锁，并且也会在等待一段时间之后一直循环尝试加锁，阻塞住，知道第一个服务实例释放锁。对于不同的服务实例尝试会获取一把锁，也和上面的逻辑类似，都是这样实现了锁的互斥。
紧接着，我们来看看锁释放的逻辑，其实也很简单，调用了 lock.unlock()方法，跟着代码走流程发现，也是异步调用了一段 lua 脚本，lua 脚本，应该就比较清晰，也就是通过判断锁 key 是否存在，如果不存在直接返回；否则就会判断当前客户端对应的唯一 key 的值是否存在，如果不存在就会返回 nil;否则，值自增-1，判断唯一 key 的值是否大于零，如果大于零，则返回 0；否则删除当前锁 key，并返回 1；返回到上一层方法，也是针对返回值进行了操作，如果返回值是 1，则会去取消之前的定时续约任务，如果失败了，则会做一些类似设置状态的操作，这一些和解锁逻辑也没有什么关系，可以不去看他。
解锁流程源码
解锁流程相对比较简单，完全就是执行解锁 Lua 脚本，无额外的代码逻辑，直接看 org.redisson.RedissonLock#unlock 代码
@Override
public void unlock() {
// 执行解锁Lua脚本，这里传入线程id，是为了保证加锁和解锁是同一个线程，避免误解锁其他线程占有的锁
Boolean opStatus = get(unlockInnerAsync(Thread.currentThread().getId()));
if (opStatus == null) {
throw new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "
+ id + " thread-id: " + Thread.currentThread().getId());
}
if (opStatus) {
cancelExpirationRenewal();
}
}
// 见org.redisson.RedissonLock#unlockInnerAsync
protected RFuture<Boolean> unlockInnerAsync(long threadId) {
return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
"if (redis.call('exists', KEYS[1]) == 0) then " +
"redis.call('publish', KEYS[2], ARGV[1]); " +
"return 1; " +
"end;" +
"if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
"return nil;" +
"end; " +
"local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
"if (counter > 0) then " +
"redis.call('pexpire', KEYS[1], ARGV[2]); " +
"return 0; " +
"else " +
"redis.call('del', KEYS[1]); " +
"redis.call('publish', KEYS[2], ARGV[1]); " +
"return 1; "+
"end; " +
"return nil;",
Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.unlockMessage, internalLockLeaseTime, getLockName(threadId));
}加锁&解锁流程串起来
上面结合 Lua 脚本和源码，分别分析了加锁流程和解锁流程。下面升级下挑战难度，模拟下多个线程争抢锁会是怎样的流程。示意图如下，比较关键的三处已用红色字体标注。
概括下整个流程
1、线程 A 和线程 B 两个线程同时争抢锁。线程 A 很幸运，最先抢到了锁。线程 B 在获取锁失败后，并未放弃希望，而是主动订阅了解锁消息，然后再尝试获取锁，顺便看看没有抢到的这把锁还有多久就过期，线程 B 就按需阻塞等锁释放。
2、线程 A 拿着锁干完了活，自觉释放了持有的锁，于此同时广播了解锁消息，通知其他抢锁的线程再来枪；
3、解锁消息的监听者 LockPubSub 收到消息后，释放自己持有的信号量；线程 B 就瞬间从阻塞中被唤醒了，接着再抢锁，这次终于抢到锁了！后面再按部就班，干完活，解锁
需要特别注意的是，RedissonLock 同样没有解决 节点挂掉的时候，存在丢失锁的风险的问题。而现实情况是有一些场景无法容忍的，所以 Redisson 提供了实现了 redlock 算法的 RedissonRedLock，RedissonRedLock 真正解决了单点失败的问题，代价是需要额外的为 RedissonRedLock 搭建 Redis 环境。
所以，如果业务场景可以容忍这种小概率的错误，则推荐使用 RedissonLock， 如果无法容忍，则推荐使用 RedissonRedLock。
redlock 算法
在分布式版本的算法里我们假设我们有 N 个 Redis master 节点，这些节点都是完全独立的，我们不用任何复制或者其他隐含的分布式协调机制。之前我们已经描述了在 Redis 单实例下怎么安全地获取和释放锁。我们确保将在每（N)个实例上使用此方法获取和释放锁。在我们的例子里面我们把 N 设成 5，这是一个比较合理的设置，所以我们需要在 5 台机器上面或者 5 台虚拟机上面运行这些实例，这样保证他们不会同时都宕掉。为了取到锁，客户端应该执行以下操作:
1.获取当前 Unix 时间，以毫秒为单位。2.依次尝试从 5 个实例，使用相同的 key 和具有唯一性的 value（例如 UUID）获取锁。当向 Redis 请求获取锁时，客户端应该设置一个尝试从某个 Reids 实例获取锁的最大等待时间（超过这个时间，则立马询问下一个实例），这个超时时间应该小于锁的失效时间。例如你的锁自动失效时间为 10 秒，则超时时间应该在 5-50 毫秒之间。这样可以避免服务器端 Redis 已经挂掉的情况下，客户端还在死死地等待响应结果。如果服务器端没有在规定时间内响应，客户端应该尽快尝试去另外一个 Redis 实例请求获取锁。3.客户端使用当前时间减去开始获取锁时间（步骤 1 记录的时间）就得到获取锁消耗的时间。当且仅当从大多数（N/2+1，这里是 3 个节点）的 Redis 节点都取到锁，并且使用的总耗时小于锁失效时间时，锁才算获取成功。4.如果取到了锁，key 的真正有效时间 = 有效时间（获取锁时设置的 key 的自动超时时间） - 获取锁的总耗时（询问各个 Redis 实例的总耗时之和）（步骤 3 计算的结果）。5.如果因为某些原因，最终获取锁失败（即没有在至少 “N/2+1 ”个 Redis 实例取到锁或者“获取锁的总耗时”超过了“有效时间”），客户端应该在所有的 Redis 实例上进行解锁（即便某些 Redis 实例根本就没有加锁成功，这样可以防止某些节点获取到锁但是客户端没有得到响应而导致接下来的一段时间不能被重新获取锁）。
用 Redisson 实现分布式锁(红锁 RedissonRedLock)及源码分析（实现三）
这里以三个单机模式为例，需要特别注意的是他们完全互相独立，不存在主从复制或者其他集群协调机制。
Config config1 = new Config();
config1.useSingleServer().setAddress("redis://172.0.0.1:5378").setPassword("a123456").setDatabase(0);
RedissonClient redissonClient1 = Redisson.create(config1);
Config config2 = new Config();
config2.useSingleServer().setAddress("redis://172.0.0.1:5379").setPassword("a123456").setDatabase(0);
RedissonClient redissonClient2 = Redisson.create(config2);
Config config3 = new Config();
config3.useSingleServer().setAddress("redis://172.0.0.1:5380").setPassword("a123456").setDatabase(0);
RedissonClient redissonClient3 = Redisson.create(config3);
/**
* 获取多个 RLock 对象
*/
RLock lock1 = redissonClient1.getLock(lockKey);
RLock lock2 = redissonClient2.getLock(lockKey);
RLock lock3 = redissonClient3.getLock(lockKey);
/**
* 根据多个 RLock 对象构建 RedissonRedLock （最核心的差别就在这里）
*/
RedissonRedLock redLock = new RedissonRedLock(lock1, lock2, lock3);
try {
/**
* 4.尝试获取锁
* waitTimeout 尝试获取锁的最大等待时间，超过这个值，则认为获取锁失败
* leaseTime 锁的持有时间,超过这个时间锁会自动失效（值应设置为大于业务处理的时间，确保在锁有效期内业务能处理完）
*/
boolean res = redLock.tryLock((long)waitTimeout, (long)leaseTime, TimeUnit.SECONDS);
if (res) {
//成功获得锁，在这里处理业务
}
} catch (Exception e) {
throw new RuntimeException("aquire lock fail");
}finally{
//无论如何, 最后都要解锁
redLock.unlock();
}最核心的变化就是需要构建多个 RLock ,然后根据多个 RLock 构建成一个 RedissonRedLock，因为 redLock 算法是建立在多个互相独立的 Redis 环境之上的（为了区分可以叫为 Redission node），Redission node 节点既可以是单机模式(single)，也可以是主从模式(master/salve)，哨兵模式(sentinal)，或者集群模式(cluster)。这就意味着，不能跟以往这样只搭建 1 个 cluster、或 1 个 sentinel 集群，或是 1 套主从架构就了事了，需要为 RedissonRedLock 额外搭建多几套独立的 Redission 节点。 比如可以搭建 3 个 或者 5 个 Redission 节点，具体可看视资源及业务情况而定。
下图是一个利用多个 Redission node 最终 组成 RedLock 分布式锁的例子，需要特别注意的是每个 Redission node 是互相独立的，不存在任何复制或者其他隐含的分布式协调机制。
Redisson 实现 redlock 算法源码分析（RedLock）
加锁核心代码
org.redisson.RedissonMultiLock#tryLock
public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
long newLeaseTime = -1;
if (leaseTime != -1) {
newLeaseTime = unit.toMillis(waitTime)*2;
}
long time = System.currentTimeMillis();
long remainTime = -1;
if (waitTime != -1) {
remainTime = unit.toMillis(waitTime);
}
long lockWaitTime = calcLockWaitTime(remainTime);
/**
* 1. 允许加锁失败节点个数限制（N-(N/2+1)）
*/
int failedLocksLimit = failedLocksLimit();
/**
* 2. 遍历所有节点通过EVAL命令执行lua加锁
*/
List<RLock> acquiredLocks = new ArrayList<>(locks.size());
for (ListIterator<RLock> iterator = locks.listIterator(); iterator.hasNext();) {
RLock lock = iterator.next();
boolean lockAcquired;
/**
* 3.对节点尝试加锁
*/
try {
if (waitTime == -1 && leaseTime == -1) {
lockAcquired = lock.tryLock();
} else {
long awaitTime = Math.min(lockWaitTime, remainTime);
lockAcquired = lock.tryLock(awaitTime, newLeaseTime, TimeUnit.MILLISECONDS);
}
} catch (RedisResponseTimeoutException e) {
// 如果抛出这类异常，为了防止加锁成功，但是响应失败，需要解锁所有节点
unlockInner(Arrays.asList(lock));
lockAcquired = false;
} catch (Exception e) {
// 抛出异常表示获取锁失败
lockAcquired = false;
}
if (lockAcquired) {
/**
*4. 如果获取到锁则添加到已获取锁集合中
*/
acquiredLocks.add(lock);
} else {
/**
* 5. 计算已经申请锁失败的节点是否已经到达 允许加锁失败节点个数限制 （N-(N/2+1)）
* 如果已经到达， 就认定最终申请锁失败，则没有必要继续从后面的节点申请了
* 因为 Redlock 算法要求至少N/2+1 个节点都加锁成功，才算最终的锁申请成功
*/
if (locks.size() - acquiredLocks.size() == failedLocksLimit()) {
break;
}
if (failedLocksLimit == 0) {
unlockInner(acquiredLocks);
if (waitTime == -1 && leaseTime == -1) {
return false;
}
failedLocksLimit = failedLocksLimit();
acquiredLocks.clear();
// reset iterator
while (iterator.hasPrevious()) {
iterator.previous();
}
} else {
failedLocksLimit--;
}
}
/**
* 6.计算 目前从各个节点获取锁已经消耗的总时间，如果已经等于最大等待时间，则认定最终申请锁失败，返回false
*/
if (remainTime != -1) {
remainTime -= System.currentTimeMillis() - time;
time = System.currentTimeMillis();
if (remainTime <= 0) {
unlockInner(acquiredLocks);
return false;
}
}
}
if (leaseTime != -1) {
List<RFuture<Boolean>> futures = new ArrayList<>(acquiredLocks.size());
for (RLock rLock : acquiredLocks) {
RFuture<Boolean> future = ((RedissonLock) rLock).expireAsync(unit.toMillis(leaseTime), TimeUnit.MILLISECONDS);
futures.add(future);
}
for (RFuture<Boolean> rFuture : futures) {
rFuture.syncUninterruptibly();
}
}
/**
* 7.如果逻辑正常执行完则认为最终申请锁成功，返回true
*/
return true;
}
评论 (0)