分布式应用进行逻辑处理时经常会遇到并发问题，这个时候就要使用到分布式锁来限制程序的并发执行。分布式锁的实现方式有很多种，ZooKeeper、Redis还有MySql的排他锁等等，网上相关的文章也是层出不穷。怎么说呢，个人感觉：凡是跟分布式沾点边的东西，就很难找到一种完美的解决方案。各有优缺点吧，我们在选型的时候吧跟找对象一样，适合自己的才是最好的。
我刚接触redis不久的时候，有一个需求是写一个过滤重复请求的AOP。于是就有了下面这段代码，今天偶然间review了下，虽然当时注意到了一些细节，但还是有满多槽点的。当然这段代码已经上线一年了，而我也不在那家公司了（手动狗头）。我想通过这段代码，和一些刚刚接触redis的朋友分享一下，设计分布式锁应该注意哪些问题。
从一段线上代码思考如何设计redis锁
private static final ScheduledExecutorService EXECUTOR_SERVICE = Executors.newScheduledThreadPool(10);
private static final String DELIMITER = "|";
@Autowired
private StringRedisTemplate template;
/**
* 延迟unlock
*
* @param lockKey key
* @param uuid client(最好是唯一键的)
* @param timeout 超时时间
* @param unit 时间单位
*/
public boolean lock(String lockKey, final String uuid, long timeout, final TimeUnit unit) {
final long milliseconds = Expiration.from(timeout, unit).getExpirationTimeInMilliseconds();
final long currentTimeMillis = System.currentTimeMillis();
boolean success = template.opsForValue().setIfAbsent(lockKey, ( currentTimeMillis + milliseconds) + DELIMITER + uuid);
if (success) {
//上锁成功
template.expire(lockKey, timeout, unit);
} else {
String oldVal = template.opsForValue().getAndSet(lockKey, (currentTimeMillis + milliseconds) + DELIMITER + uuid);
final String[] oldValues = oldVal.split(Pattern.quote(DELIMITER));
if (Long.parseLong(oldValues[0]) + 1 <= currentTimeMillis) {
//临界区间，判断锁是否失效，失效重新获取锁
template.expire(lockKey, timeout, unit);
return true;
}
}
return success;
}
/**
* 延迟unlock
*
* @param lockKey key
* @param uuid client(最好是唯一键的)
* @param delayTime 延迟时间
* @param unit 时间单位
*/
public void unlock(final String lockKey, final String uuid, long delayTime, TimeUnit unit) {
if (StringUtils.isEmpty(lockKey)) {
return;
}
if (delayTime <= 0) {
doUnlock(lockKey, uuid);
} else {
EXECUTOR_SERVICE.schedule(() -> doUnlock(lockKey, uuid), delayTime, unit);
}
}
/**
* @param lockKey key
* @param uuid client(最好是唯一键的)
*/
private void doUnlock(final String lockKey, final String uuid) {
String val = template.opsForValue().get(lockKey);
final String[] values = val.split(Pattern.quote(DELIMITER));
if (values.length <= 0) {
return;
}
//确保当前线程占有的锁不会被其它线程释放
if (uuid.equals(values[1])) {
template.delete(lockKey);
}
}
1.1 基本设计思路
image
- 获取锁
获取锁实际就是在redis里面占一个“坑”，当一个线程先抢到了这个“坑”，下一个需要进这个坑位的线程就在外边等着。这里我们会使用setnx(set if not exists)指令，对应代码中的方法是template.opsForValue().setIfAbsent(key，value)。意思是我拿一个key看下redis里面有没有，如果没有，就创建一个把value设置进去，如果有了就拜拜。这里返回true我们就认为线程是第一个访问的，抢到了redis锁，返回false说明前面已经有人再用了。
正常情况我们在获取锁后，执行业务逻辑，然后在释放锁。如果执行业务逻辑时发生了异常，可能就走不到释放锁的操作，会造成死锁，消耗客户端资源。所以在拿到锁以后，我们可以通过expire设置一个过期时间，即使出现异常也能保证锁在有效时间后会自动失效，最终无效的key被redis回收。
- 释放锁
释放锁执行del指令就可以了，因为前面我们设置了过期时间的缘故，这里我们可以写一个定时job，等到失效时间过了来执行删除操作即可
1.2 缺陷
上面提到了我们通过setnx和expire指令来获取锁，通过del指令来释放锁，这是我们设计redis锁的基本思路，但同样存在一些问题。
-
setnx和setex并不是原子性操作
如果在 setnx 和 expire 之间服务器进程突然挂掉了,会导致 expire 得不到执行，也会造成死锁。这种问题的根源就在于 setnx 和 expire 是两条指令而不是原子指令。
- 删除也不是绝对安全的
线程A在获取锁后，执行业务逻辑，但是业务逻辑执行的时间太长了，锁已经失效了。这个时候线程B重新持有了锁，开始执行业务逻辑。A线程开始执行释放锁操作，把B的锁释放了。
1.3 解决思路
上面两个问题是我在写代码之前就有了解到的，我来聊一下体现在代码里的解决思路。
-
setnx和setex并不是原子性操作
这个问题其实在Redis2.6.12之前都是通过lua脚本解决的。Redis 2.6.12版本中作者加入了set 指令的扩展参数，使得 setnx 和expire指令可以一起执行，彻底解决了分布式锁的乱象。
那么大家肯定会问了，说好的lua脚本呢？你的代码里怎么没有呢？
说到这里，我不得不说一下我的心酸史。起初我也是网上找了一个lua脚本的demo，kuangkuang就给干上去了，测了下也没啥问题，当时还觉得自己挺吊。等到上线的时候，接口跌停了。我们当时的redis是在k8s里的，跟测试环境也不一样。版本比较低，不支持lua脚本，尼玛我当时就尿了，回滚了代码。后来查阅资料才知道，从 Redis 2.6.0后才支持 lua 脚本的执行 。所以说朋友们，我们在项目了引入什么新鲜东西的时候，一定要注意实际的生产环境呀！！ 还有个教训就是和redis有关的操作都try catch下吧，等你们redis出故障的时候，你会来感谢我的。
不能用lua脚本，保证不了原子性，我想了一个补偿方案，曲线救国。假设线程A在设置expire的时候失败了，线程B进来会抢不到锁。这个时候如果我们能知道线程A是什么时候访问的，自己来判断下它是否过期，如果过期了，我们就认为这个锁是无效的，把它给B线程用就好了。所以在代码中，我们的value记录了锁的过期时间。下一个线程进来时通过getAndSet获取上一次的value值拿来做比对，在将自己的value写入redis中。getAndSet是一个原子操作,就这样完成了替换。当然在对B线程设置过期时间时依然会存在原子性问题，那就下一次补偿吧，我也没招了。后来我们把redis从容器中拿了出来，升级了版本，就不存在这个问题了。
- 删除也不是绝对安全的
解决这个问题，首先要保证锁的唯一性。就是获取的锁和释放的锁应该是独有的，所以在代码中我们加入UUID作为锁的标识。在删除时我们去对比下UUID，如果匹配上了在进行删除。但是这里又会出现另外一个问题：获取和删除不是原子的呀！所以释放锁，一定要使用lua脚本。保证其原子性。
哦，兄弟们可能又要问了，为啥我的代码里没有。哈哈版本不支持，然后因为我做的需求是过滤重复请求，对于重复的请求挡掉就可以了，不需要让他们阻塞，所以当时就没有对释放锁这块进行处理。我补上好吧!
//释放锁lua脚本
private static final String RELEASE_LOCK_LUA_SCRIPT = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
private void doUnlock2(final String lockKey, final String uuid) {
// 指定 lua 脚本，并且指定返回值类型
DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>(RELEASE_LOCK_LUA_SCRIPT,Long.class);
try {
template.execute(redisScript, Collections.singletonList(lockKey),uuid);
} catch (Exception e) {
e.printStackTrace();
}
}
在看了我的当时的心路历程后，相信兄弟们对redis锁已经有点概念了。其实上面考虑的情况都是单机版Redis存在的问题，稍加注意都可以解决，集群情况下仍存在的隐患。其实说句实话，很多东西是把非常极端的情况考虑了进去，至少这垃圾代码在线上跑了一年多倒是没啥事故，但我们做技术还是要严谨些，考虑的全面些。
设计Redis锁你需要注意
2.1 必须设置过期时间
锁必须要设置一个过期时间。否则的话，当一个客户端获取锁成功之后，假如它崩溃了，或者由于发生了网络分割（network partition）导致它再也无法和Redis节点通信了，那么它就会一直持有这个锁，而其它客户端永远无法获得锁了。
2.2 执行exprie之前客户端崩溃了怎么办
要看下redis的版本，2.6.0以上的版本就可以通过lua脚本合并setnx和exprie解决。2.6.12以后set命令增加了EX，PX，NX和XX选项支持了过期时间的设置。
image
2.3 保证value值的唯一性
设置一个随机字符串是很有必要的，它保证了一个客户端释放的锁必须是自己持有的那个锁。假如获取锁时SET的不是一个随机字符串，而是一个固定值，那么可能某个客户端因为阻塞等原因，可能会误删其他客户端正在持有的锁。
2.4 释放锁必须使用lua脚本
释放锁的操作必须使用Lua脚本来实现。释放锁其实包含三步操作：GET、判断和DEL，用Lua脚本来实现能保证这三步的原子性。否则，如果把这三步操作放到客户端逻辑中去执行的话，就有可能发生与前面第三个问题类似的执行序列：
- 客户端1获取锁成功。
- 客户端1访问共享资源。
- 客户端1为了释放锁，先执行
GET操作获取随机字符串的值。 - 客户端1判断随机字符串的值，与预期的值相等。
- 客户端1由于某个原因阻塞住了很长时间。
- 过期时间到了，锁自动释放了。
- 客户端2获取到了对应同一个资源的锁。
- 客户端1从阻塞中恢复过来，执行
DEL操纵，释放掉了客户端2持有的锁。
实际上，在上述第三个问题和第四个问题的分析中，如果不是客户端阻塞住了，而是出现了大的网络延迟，也有可能导致类似的执行序列发生
2.5 尴尬的超时时间设置问题
超时设置成多少合适呢？如果设置太短的话，锁就有可能在客户端完成对于共享资源的访问之前过期，从而失去保护；如果设置太长的话，一旦某个持有锁的客户端释放锁失败，那么就会导致所有其它客户端都无法获取锁，从而长时间内无法正常工作。看来真是个两难的问题，个人不建议使用redis锁处理太复杂的业务逻辑。
2.6 如果Sentinel集群的主节点挂了怎么办？
在 Sentinel 集群中，Master节点挂掉时，Slave节点会取而代之，但由于Redis的主从复制（replication）是异步的，这可能导致在failover过程中丧失锁的安全性。
- 客户端1从Master获取了锁。
- Master宕机了，存储锁的key还没有来得及同步到Slave上。
- Slave升级为Master。
- 客户端2从新的Master获取到了对应同一个资源的锁。
针对这个问题，antirez设计了Redlock算法,用来解决Redis分布式锁存在的一致性问题。不过引入Redlock也会存在需要创建多实例的成本问题，如果业务并不是很需要高可用，可以忽略failover引起的问题。
下一篇我会介绍一下Redlock算法以及优秀的开源解决方案Redission，我知道兄弟们可能对setnx+Lua脚本的代码忍不了了，Redission会帮我们解决这个问题的，敬请期待吧～
网友评论