基于redis实现分布式锁的几种方案以及不足
方案1：setnx 方案（不建议使用）
redis 提供 setnx 命令，是「SET if Not eXists」的缩写，只有不存在时才会设置返回1，否则返回0，如下：
127.0.0.1:6379> setnx javabk.cn 1
(integer) 1
127.0.0.1:6379> setnx javabk.cn 1
(integer) 0
实际设置成功时，表示获取到锁，一般会马上通过 expire 设置过期时间，避免处理业务时没有及时删除导致后面的请求都获取不到锁，具体例子如下：
public class DistributedLockDemoTest {
Jedis jedis = new Jedis("127.0.0.1",6379);
@Test
public void setnxAndExpire() {
String myId = UUID.randomUUID().toString();
String key = "javabk.cn";
//1. 通过 setnx 抢锁
long result = jedis.setnx(key, myId);
//2. 结果判断
if (result == 1) {//成功获取锁
try {
//3. 设置过期，避免死锁
jedis.expire(key, 30);//30 seconds expired
//4. 业务处理....
} finally {
//5. 释放锁（这里可以优化成：将判断+删除写成lua脚本进行删除）
if (myId.equals(jedis.get(key))) {//判断value是自身设置才删除
jedis.del(key);
}
}
} else { //获取锁失败
// ....
}
}
}
该方案存在的问题
- 存在死锁的可能：如果在setnx设置完成后(代码 [1] 地方)，再通过 expire 设置（代码 [3]）之前程序重启或者挂了，那么这个key将无法解锁。核心原因是setnx + expire 是通过两次网络进行发送到redis执行的，无法保证其原子性。该问题的解决方案参考方案2 和 方案3
- 锁在持有期间过期：假如处理逻辑时间（加期间 FULL GC 时间）超过key的过期时间，那么锁可能别其他客户端获取，同时处理相关逻辑，可能影响业务结果，这个超时问题没很好的彻底解决办法，不过可以通过一些策略来降低发生的概率：尽量控制处理时间(比如查询接口限制超时时间等)+ 预留的 FULL GC 时间 小于key过期时间，同时考虑后台线程对key进行定时续期（有人称 watch dog，redisson框架有支持）。
方案2：set扩展命令
针对方案1存在的问题，在redis版本 >=2.8 ，针对set 命令进行扩展来解决这个setnx + expire 的原子性问题。命令如下：
SET key value [EX seconds] [PX milliseconds] [NX|XX]
其中 EX 表示秒，PX 表示毫秒，NX 表示不存在才设置，XX 表示存在才设置。将命令其实就是将 setnx + expire 2个命令合并成1个，保证了原子性。命令例子如下：
127.0.0.1:6379> set javabk.cn 1 ex 30 nx //不存在时设置成功返回OK
OK
127.0.0.1:6379> ttl javabk.cn
(integer) 24
127.0.0.1:6379> set javabk.cn 1 ex 30 nx //存在时，设置不成功，返回空
(nil)
代码例子：
public void setExtendCommand() {
String myId = UUID.randomUUID().toString();
String key = "javabk.cn";
//1. 通过 setnx 抢锁
String result = jedis.set(key, myId, SetParams.setParams().ex(30).nx());
System.out.println("result is:" + result);
//2. 结果判断
if ("OK".equals(result)) {//成功获取锁
try {
//3. 业务处理....
} finally {
//4.释放锁
unLockAfterCompareWithLua(key, myId);
}
} else { //获取锁失败
// ....
}
}
public boolean unLockAfterCompareWithLua(String key, String value) {
String luaSrcipt = "if redis.call('get',KEYS[1]) == ARGV[1] then\n" +
"redis.call('del',KEYS[1])\n" +
"return 1 \n" +
"else\n" +
"return 0\n" +
"end";
List<String> placeHolderKeys = Lists.newArrayList(key);
List<String> placeHolderValues = Lists.newArrayList(value);
Object result = jedis.eval(luaSrcipt, placeHolderKeys, placeHolderValues);
if (result != null && "1".equals(result.toString())) {
return true;
}
return false;
}
方案3：通过lua脚本打包 setnx + expire 命令
redis可以通过lua脚本打包多个命令进行执行，保证其执行原子性，可以解决 setnx + expire 原子性执行问题。其实多个命令执行的原子性问题都可以通过将其打包成lua脚本来保证原子性执行。
代码例子：
public void setNxAndExpireWithLua() {
String myId = UUID.randomUUID().toString();
String key = "javabk.cn";
//放到服务端执行的lua脚本
String luaSrcipt = "if redis.call('setnx',KEYS[1],ARGV[1]) == 1 then\n" +
"redis.call('expire',KEYS[1],ARGV[2])\n" +
"return 1 \n" +
"else\n" +
"return 0\n" +
"end";
//1. 通过 setnx 抢锁
List<String> placeHolderKeys = Lists.newArrayList(key);
List<String> placeHolderValues = Lists.newArrayList(myId, "30");
Object result = jedis.eval(luaSrcipt, placeHolderKeys, placeHolderValues);//核心改动
System.out.println("result is:" + result);
//2. 结果判断
if ("1".equals(result)) {//成功获取锁（跟lua脚本的返回1对应）
try {
//3. 业务处理....
} finally {
//4. 释放锁
unLockAfterCompareWithLua(key, myId);
}
} else { //获取锁失败
// ....
}
}
public boolean unLockAfterCompareWithLua(String key, String value) {
String luaSrcipt = "if redis.call('get',KEYS[1]) == ARGV[1] then\n" +
"redis.call('del',KEYS[1])\n" +
"return 1 \n" +
"else\n" +
"return 0\n" +
"end";
List<String> placeHolderKeys = Lists.newArrayList(key);
List<String> placeHolderValues = Lists.newArrayList(value);
Object result = jedis.eval(luaSrcipt, placeHolderKeys, placeHolderValues);
if (result != null && "1".equals(result.toString())) {
return true;
}
return false;
}
其实通过方案2和方案3，可以解决大部分业务场景，如果有些业务场景需要锁的可重入性，那么可以参考[可重入性]
方案4 基于redisson(推荐使用)
redisson 是基于一个 redis java client，底层实现做了很多封装，比如分布式锁、读写锁等等，具体请看 官网
核心代码：
public class RedisLockWithRedisson {
RedissonClient redissonClient = null;
public RedisLockWithRedisson(String host, int port) {
Config config = new Config();
config.useSingleServer().setAddress("redis://" + host + ":" + port);
config.setLockWatchdogTimeout(10 * 1000);//10s. 覆盖 watch log 默认30s 超时的配置
redissonClient = Redisson.create(config);
}
public boolean lockWithWatchDog(String key, int waitSecond) throws Exception {
RLock lock = redissonClient.getLock(key);
//尝试加锁，第一个参数表示最多等待多少秒。启动 watch log 续期
boolean locked = lock.tryLock(waitSecond, TimeUnit.SECONDS);
if (locked) {
System.out.println("成功获取锁.key:" + key);
System.out.println("业务处理...");
//业务处理
for (int i = 0; i < 10; i++) {
System.out.println("业务处理中，剩余时间：" + lock.remainTimeToLive());
Thread.sleep(1500);
}
System.out.println("处理业务完成后，锁是否存在:" + lock.isLocked());
lock.unlock();
System.out.println("解锁后，锁是否存在:" + lock.isLocked());
return true;
} else {
System.out.println("获取锁失败在等待: " + waitSecond+ "秒,key:" + key);
return false;
}
}
}
说明：redisson 对比上面几个方案，其实实现是类似的，只不过做了大量的封装，使用非常简单，而且内部增加了 watch dog 续期机制。我们看看上面的最核心代码：lock.tryLock(waitSecond, TimeUnit.SECONDS) ，其底层实现就是一个lua脚本，如下：
if (redis.call('exists', KEYS[1]) == 0) then
redis.call('hincrby', 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]);
通过脚本可发现，其通过 hash结构来支持锁的可重入性，hash key 是给每个线程（客户端）分配的唯一ID，hash value 是同个线程重复成功加锁的次数。加锁成功后，开启一个定时任务每隔一段时间继续续期，默认是过期时间是30秒，每隔 1/3 超时时间进行1次续期，上面例子覆盖默认超时时间，改成10秒。例子中特意实现业务处理时间超过过期时间，但是由于续期机制，保证处理期间不过期。
测试代码：
@Test
public void testLockWithWhatDogAndWait() throws Exception {
String key = "javabk.cn";
for (int i = 0; i < 2; i++) {
Runnable runnable = () -> {
try {
lockWithRedisson.lockWithWatchDog(key, 10);
} catch (Exception e) {
e.printStackTrace();
}
};
executor.submit(runnable);
}
Thread.currentThread().join();
}
测试结果如下：由于设置watch dog 超时时间10秒，所以 3秒进行1次续期（1/3 * 10），所以从6970 ttl 变成 8834 主要是由于续期带来的
成功获取锁.key:javabk.cn
业务处理...
业务处理中，剩余时间：9993
业务处理中，剩余时间：8480
业务处理中，剩余时间：6970
业务处理中，剩余时间：8834
业务处理中，剩余时间：7323
业务处理中，剩余时间：9212
业务处理中，剩余时间：7701
获取锁失败在等待: 10秒,key:javabk.cn
业务处理中，剩余时间：9588
业务处理中，剩余时间：8082
业务处理中，剩余时间：9971
处理业务完成后，锁是否存在:true
解锁后，锁是否存在:false
可重入性
可重入性主要是表示同一个线程可以对一个锁重复加锁和释放锁，类似java的 ReentrantLock。因为是针对同一个线程，可以考虑将重入锁的统计和方式记录到本地内存，提升性能。下面例子是基于内存来实现锁的可重入性，也可以通过 redis 的hash 结构来实现，hash key 为给线程分配的唯一ID，value 是重入次数，Redisson 实现锁重入就是这样实现的，可以参考上面的例子。
基于内存实现锁的可重入性核心代码入下：
private ThreadLocal<Map<String, Integer>> threadLocks = new ThreadLocal<>();
private Map<String, Integer> getCurrentThreadLocks() {
Map<String, Integer> refs = threadLocks.get();
if (refs != null) {
return refs;
}
threadLocks.set(new HashMap<>());
return threadLocks.get();
}
//锁-可重入性
public boolean lockReentrant(String key, String value, int expireSecond) {
Map<String, Integer> currentThreadLocks = getCurrentThreadLocks();
Integer thisLockCount = currentThreadLocks.get(key);
if (thisLockCount != null) {//如果本地有锁，说明已经在redis上加过锁，本地内存进行累计即可
currentThreadLocks.put(key, thisLockCount + 1);
return true;
}
boolean ok = this.lock(key, value, expireSecond);
if (!ok) {
return false;
}
currentThreadLocks.put(key, 1);
return true;
}
//解锁-可重入性
public boolean unlock(String key, String value) {
Map<String, Integer> currentThreadLocks = getCurrentThreadLocks();
Integer thisLockCount = currentThreadLocks.get(key);
if (thisLockCount == null) {
return true;
}
thisLockCount = thisLockCount - 1;
if (thisLockCount > 0) {
currentThreadLocks.put(key, thisLockCount);
} else {//如果没有锁，需要对redis的锁进行删除
currentThreadLocks.remove(key);
unLock(key, value);
}
return true;
}
redis分布式锁在集群的问题
如果redis 不仅仅是一个节点，比如搭建主从集群，那么可能存在这样的问题：如果在主节点成功加锁，但是在同步给从节点的时候还没完成将命令同步给从节点时，主节点就挂了，从节点变成了主节点，这时候这个锁在新主节点就不存在，导致新的客户端就可以加锁成功，这样可能导致同一把锁可能被两个客户端持有。其实这种case出现的概率比较低，而且持续时间很短，大部分实际的场景都是可以接受的。不过出于学习，可以了解其解决方案，目前解决方案有 RedLock算法。
RedLock 算法
核心原理主要是：需要提供多个 redis 节点，在加锁时，需向过半节点发送加锁命令，只要过半节点加锁成功，就认为加锁成功，释放时，需要将所有节点发送释放锁命令。但是可能存在以下问题：
-
加锁的主节点最好不要建立从节点，否则也会出现多个客户端同时获取同一把锁的可能。
比如：3对主从节点，客户端分别给主1，主2加锁成功，主3加锁失败（比如网络抖动），由于加锁数量超过1半，所以认为获取锁，这时候还没释放锁，主2挂了failover到主2的从节点，但是主2的加锁命令还没同步到从节点，导致从节点成为新的主节点时没有加锁信息，这时候客户端B来加锁，成功在主2加锁，同时在主3加锁成功，这时候就有2个客户端同时获取锁。虽然出现这种case的可能性比较小，但是理论上是存在这个问题。
没有开启AOF或者AOF不是每条flush(一般开启也是N秒才flush)，在服务挂了重启存在多个客户端获取锁。比如：3个主节点，客户端1分别给主1，主2 加锁成功，主3加锁失败，这时候主2挂了重启，这时候客户端1还没解锁，导致客户端1的加锁数据丢了，然后客户端2这时候在主2+主3加锁成功，这时候就2个客户端同时获取锁。
总结：虽然redis基于RedLock算法可以解决redis主从切换导致的分布式锁的问题，但是其限制条件多，性能下降，导致用不上 redis的高性能特性，如果业务真的需要强一致（任何情况都不能出现多个客户端同时获取锁），那建议不要用redis来实现分布式锁，可以考虑使用 zookeeper 实现分布式锁。不过最终还是得结合业务选择最合适的方案，没有最完美的通用方案，只有最合适自己业务的方案。
附录
本文章的代码：点击传送门
本文由mdnice多平台发布
