分布式锁没那么难，手把手教你实现 Redis 分布锁！|保姆级教程

书接上文

上篇文章「MySQL 可重复读，差点就让我背上了一个 P0 事故！」发布以后，收到不少小伙伴们的留言，从中又学习到不少，总结一下。

上篇文章可能举得例子有点不恰当，致使有些小伙伴没看懂为何余额会变负。

此次咱们举得实际一点，仍是上篇文章 account 表，假设 id=1，balance=1000，不过此次咱们扣款 1000，两个事务的时序图以下:

此次使用两个命令窗口真实执行一把：

注意事务 2，③处查询到 id=1，balance=1000，可是实际上因为此时事务 1 已经提交，最新结果如②处所示 id=1，balance=900。

原本 Java 代码层会作一层余额判断：

if (balance - amount < 0) {
  throw new XXException("余额不足，扣减失败");
}
复制代码

可是此时因为 ③ 处使用快照读，读到是个旧值，未读到最新值，致使这层校验失效，从而代码继续往下运行，执行了数据更新。

更新语句又采用以下写法:

UPDATE account set balance=balance-1000 WHERE id =1;
复制代码

这条更新语句又必须是在这条记录的最新值的基础作更新，更新语句执行结束，这条记录就变成了 id=1，balance=-1000。

以前有朋友疑惑 t12 更新以后，再次进行快照读，结果会是多少。

上图执行结果 ④ 能够看到结果为 id=1,balance=-1000,能够看到已经查询最新的结果记录。

这行数据最新版本因为是事务 2 本身更新的，自身事务更新永远对本身可见。

另外此次问题上本质上由于 Java 层与数据库层数据不一致致使，有的朋友留言提出，能够在更新余额时加一层判断：

UPDATE account set balance=balance-1000 WHERE id =1 and balance>0;
复制代码

而后更新完成，Java 层判断更新有效行数是否大于 0。这种作法确实能规避这个问题。

最后这位朋友留言总结的挺好，粘贴一下：

先赞后看，微信搜索「程序通事」，关注就完事了

手撸分布式锁

如今切回正文，这篇文章原本是准备写下 Mysql 查询左匹配的问题，可是还没研究出来。那就先写下最近在鼓捣一个东西，使用 Redis 实现可重入分布锁。

看到这里，有的朋友可能会提出来使用 redisson 不香吗，为何还要本身实现？

哎，redisson 真的很香，可是现有项目中没办法使用，只好本身手撸一个可重入的分布式锁了。

虽然用不了 redisson，可是我能够研究其源码，最后实现的可重入分布锁参考了 redisson 实现方式。

分布式锁

分布式锁特性就要在于排他性，同一时间内多个调用方加锁竞争，只能有一个调用方加锁成功。

Redis 因为内部单线程的执行，内部按照请求前后顺序执行，没有并发冲突，因此只会有一个调用方才会成功获取锁。

并且 Redis 基于内存操做，加解锁速度性能高，另外咱们还可使用集群部署加强 Redis 可用性。

加锁

使用 Redis 实现一个简单的分布式锁，很是简单，能够直接使用 SETNX 命令。

SETNX 是『SET if Not eXists』，若是不存在，才会设置，使用方法以下：

不过直接使用 SETNX 有一个缺陷，咱们没办法对其设置过时时间，若是加锁客户端宕机了，这就致使这把锁获取不了了。

有的同窗可能会提出，执行 SETNX 以后，再执行 EXPIRE 命令，主动设置过时时间，伪码以下：

var result = setnx lock "client"
if(result==1){
    // 有效期 30 s
    expire lock 30
}
复制代码

不过这样仍是存在缺陷，加锁代码并不能原子执行，若是调用加锁语句，还没来得及设置过时时间，应用就宕机了，仍是会存在锁过时不了的问题。

不过这个问题在 Redis 2.6.12 版本 就能够被完美解决。这个版本加强了 SET 命令，能够经过带上 NX,EX 命令原子执行加锁操做，解决上述问题。参数含义以下：

• EX second ：设置键的过时时间，单位为秒
• NX 当键不存在时，进行设置操做，等同与 SETNX 操做

使用 SET 命令实现分布式锁只须要一行代码：

SET lock_name anystring NX EX lock_time
复制代码

解锁

解锁相比加锁过程，就显得很是简单，只要调用 DEL 命令删除锁便可:

DEL lock_name
复制代码

不过这种方式却存在一个缺陷，可能会发生错解锁问题。

假设应用 1 加锁成功，锁超时时间为 30s。因为应用 1 业务逻辑执行时间过长，30 s 以后，锁过时自动释放。

这时应用 2 接着加锁，加锁成功，执行业务逻辑。这个期间，应用 1 终于执行结束，使用 DEL 成功释放锁。

这样就致使了应用 1 错误释放应用 2 的锁，另外锁被释放以后，其余应用可能再次加锁成功，这就可能致使业务重复执行。

为了使锁不被错误释放，咱们须要在加锁时设置随机字符串，好比 UUID。

SET lock_name uuid NX EX lock_time
复制代码

释放锁时，须要提早获取当前锁存储的值，而后与加锁时的 uuid 作比较，伪代码以下：

var value= get lock_name
if value == uuid
	// 释放锁成功
else
	// 释放锁失败
复制代码

上述代码咱们不能经过 Java 代码运行,由于没法保证上述代码原子化执行。

幸亏 Redis 2.6.0 增长执行 Lua 脚本的功能，lua 代码能够运行在 Redis 服务器的上下文中，而且整个操做将会被当成一个总体执行，中间不会被其余命令插入。

这就保证了脚本将会以原子性的方式执行，当某个脚本正在运行的时候，不会有其余脚本或 Redis 命令被执行。在其余的别的客户端看来，执行脚本的效果，要么是不可见的，要么就是已完成的。

EVAL 与 EVALSHA

EVAL

Redis 可使用 EVAL 执行 LUA 脚本，而咱们能够在 LUA 脚本中执行判断求值逻辑。EVAL 执行方式以下：

EVAL script numkeys key [key ...] arg [arg ...]
复制代码

numkeys 参数用于建明参数，即后面 key 数组的个数。

key [key ...] 表明须要在脚本中用到的全部 Redis key，在 Lua 脚本使用使用数组的方式访问 key，相似以下 KEYS[1] ， KEYS[2]。注意 Lua 数组起始位置与 Java 不一样，Lua 数组是从 1 开始。

命令最后，是一些附加参数，能够用来当作 Redis Key 值存储的 Value 值，使用方式如 KEYS 变量同样，相似以下：ARGV[1] 、 ARGV[2] 。

用一个简单例子运行一下 EVAL 命令：

eval "return {KEYS[1],KEYS[2],ARGV[1],ARGV[2],ARGV[3]}" 2 key1 key2 first second third
复制代码

运行效果以下：

能够看到 KEYS 与 ARGVS内部数组能够不一致。

在 Lua 脚本可使用下面两个函数执行 Redis 命令：

• redis.call()
• redis.pcall()

两个函数做用法与做用彻底一致，只不过对于错误的处理方式不一致，感兴趣的小伙伴能够具体点击如下连接，查看错误处理一章。

doc.redisfans.com/script/eval…

下面咱们统一在 Lua 脚本中使用 redis.call()，执行如下命令：

eval "return redis.call('set',KEYS[1],ARGV[1])" 1 foo 楼下小黑哥
复制代码

运行效果以下：

EVALSHA

EVAL 命令每次执行时都须要发送 Lua 脚本，可是 Redis 并不会每次都会从新编译脚本。

当 Redis 第一次收到 Lua 脚本时，首先将会对 Lua 脚本进行 sha1 获取签名值，而后内部将会对其缓存起来。后续执行时，直接经过 sha1 计算事后签名值查找已经编译过的脚本，加快执行速度。

虽然 Redis 内部已经优化执行的速度，可是每次都须要发送脚本，仍是有网络传输的成本，若是脚本很大，这其中花在网络传输的时间就会相应的增长。

因此 Redis 又实现了 EVALSHA 命令，原理与 EVAL 一致。只不过 EVALSHA 只须要传入脚本通过 sha1计算事后的签名值便可，这样大大的减小了传输的字节大小，减小了网络耗时。

EVALSHA命令以下：

evalsha c686f316aaf1eb01d5a4de1b0b63cd233010e63d 1 foo 楼下小黑哥
复制代码

运行效果以下：

SCRIPT FLUSH 命令用来清除全部 Lua 脚本缓存。

能够看到，若是以前未执行过 EVAL命令，直接执行 EVALSHA 将会报错。

优化执行 EVAL

咱们能够结合使用 EVAL 与 EVALSHA，优化程序。下面就不写伪码了，以 Jedis 为例，优化代码以下：

//链接本地的 Redis 服务
Jedis jedis = new Jedis("localhost", 6379);
jedis.auth("1234qwer");

System.out.println("服务正在运行: " + jedis.ping());

String lua_script = "return redis.call('set',KEYS[1],ARGV[1])";
String lua_sha1 = DigestUtils.sha1DigestAsHex(lua_script);

try {
    Object evalsha = jedis.evalsha(lua_sha1, Lists.newArrayList("foo"), Lists.newArrayList("楼下小黑哥"));
} catch (Exception e) {
    Throwable current = e;
    while (current != null) {
        String exMessage = current.getMessage();
        // 包含 NOSCRIPT，表明该 lua 脚本从未被执行，须要先执行 eval 命令
        if (exMessage != null && exMessage.contains("NOSCRIPT")) {
            Object eval = jedis.eval(lua_script, Lists.newArrayList("foo"), Lists.newArrayList("楼下小黑哥"));
            break;
        }

    }
}
String foo = jedis.get("foo");
System.out.println(foo);
复制代码

上面的代码看起来仍是很复杂吧，不过这是使用原生 jedis 的状况下。若是咱们使用 Spring Boot 的话，那就没这么麻烦了。Spring 组件执行的 Eval 方法内部就包含上述代码的逻辑。

不过须要注意的是，若是 Spring-Boot 使用 Jedis 做为链接客户端,而且使用Redis Cluster 集群模式，须要使用 2.1.9 以上版本的spring-boot-starter-data-redis,否则执行过程当中将会抛出：

org.springframework.dao.InvalidDataAccessApiUsageException: EvalSha is not supported in cluster environment.
复制代码

详细状况能够参考这个修复的 IssueAdd support for scripting commands with Jedis Cluster

优化分布式锁

讲完 Redis 执行 LUA 脚本的相关命令，咱们来看下如何优化上面的分布式锁，使其没法释放其余应用加的锁。

如下代码基于 spring-boot 2.2.7.RELEASE 版本，Redis 底层链接使用 Jedis。

加锁的 Redis 命令以下：

SET lock_name uuid NX EX lock_time
复制代码

加锁代码以下：

/** * 非阻塞式加锁，若锁存在，直接返回 * * @param lockName 锁名称 * @param request 惟一标识，防止其余应用/线程解锁，可使用 UUID 生成 * @param leaseTime 超时时间 * @param unit 时间单位 * @return */
public Boolean tryLock(String lockName, String request, long leaseTime, TimeUnit unit) {
    // 注意该方法是在 spring-boot-starter-data-redis 2.1 版本新增长的，如果以前版本 能够执行下面的方法
    return stringRedisTemplate.opsForValue().setIfAbsent(lockName, request, leaseTime, unit);
}
复制代码

因为setIfAbsent方法是在 spring-boot-starter-data-redis 2.1 版本新增长，以前版本没法设置超时时间。若是使用以前的版本的，须要以下方法：

/** * 适用于 spring-boot-starter-data-redis 2.1 以前的版本 * * @param lockName * @param request * @param leaseTime * @param unit * @return */
public Boolean doOldTryLock(String lockName, String request, long leaseTime, TimeUnit unit) {
    Boolean result = stringRedisTemplate.execute((RedisCallback<Boolean>) connection -> {
        RedisSerializer valueSerializer = stringRedisTemplate.getValueSerializer();
        RedisSerializer keySerializer = stringRedisTemplate.getKeySerializer();

        Boolean innerResult = connection.set(keySerializer.serialize(lockName),
                valueSerializer.serialize(request),
                Expiration.from(leaseTime, unit),
                RedisStringCommands.SetOption.SET_IF_ABSENT
        );
        return innerResult;
    });
    return result;
}
复制代码

解锁须要使用 Lua 脚本:

-- 解锁代码
-- 首先判断传入的惟一标识是否与现有标识一致
-- 若是一致，释放这个锁，不然直接返回
if redis.call('get', KEYS[1]) == ARGV[1] then
   return redis.call('del', KEYS[1])
else
   return 0
end
复制代码

这段脚本将会判断传入的惟一标识是否与 Redis 存储的标示一致，若是一直，释放该锁，不然马上返回。

释放锁的方法以下：

/** * 解锁 * 若是传入应用标识与以前加锁一致，解锁成功 * 不然直接返回 * @param lockName 锁 * @param request 惟一标识 * @return */
public Boolean unlock(String lockName, String request) {
    DefaultRedisScript<Boolean> unlockScript = new DefaultRedisScript<>();
    unlockScript.setLocation(new ClassPathResource("simple_unlock.lua"));
    unlockScript.setResultType(Boolean.class);
    return stringRedisTemplate.execute(unlockScript, Lists.newArrayList(lockName), request);
}
复制代码

因为公号外链没法直接跳转，关注『程序通事』，回复分布式锁获取源代码。

Redis 分布式锁的缺陷

没法重入

因为上述加锁命令使用了 SETNX ，一旦键存在就没法再设置成功，这就致使后续同一线程内继续加锁，将会加锁失败。

若是想将 Redis 分布式锁改形成可重入的分布式锁，有两种方案：

• 本地应用使用 ThreadLocal 进行重入次数计数，加锁时加 1，解锁时减 1，当计数变为 0 释放锁
• 第二种，使用 Redis Hash 表存储可重入次数，使用 Lua 脚本加锁/解锁

第一种方案能够参考这篇文章分布式锁的实现之 redis 篇。第二个解决方案，下一篇文章就会具体来聊聊，敬请期待。

锁超时释放

假设线程 A 加锁成功，锁超时时间为 30s。因为线程 A 内部业务逻辑执行时间过长，30s 以后锁过时自动释放。

此时线程 B 成功获取到锁，进入执行内部业务逻辑。此时线程 A 还在执行执行业务，而线程 B 又进入执行这段业务逻辑，这就致使业务逻辑重复被执行。

这个问题我以为，通常因为锁的超时时间设置不当引发，能够评估下业务逻辑执行时间，在这基础上再延长一下超时时间。

若是超时时间设置合理，可是业务逻辑还有偶发的超时，我的以为须要排查下业务执行过长的问题。

若是说必定要作到业务执行期间，锁只能被一个线程占有的，那就须要增长一个守护线程，定时为即将的过时的但未释放的锁增长有效时间。

加锁成功后，同时建立一个守护线程。守护线程将会定时查看锁是否即将到期，若是锁即将过时，那就执行 EXPIRE 等命令从新设置过时时间。

说实话，若是要这么作，真的挺复杂的，感兴趣的话能够参考下 redisson watchdog 实现方式。

Redis 分布式锁集群问题

为了保证生产高可用，通常咱们会采用主从部署方式。采用这种方式，咱们能够将读写分离，主节点提供写服务，从节点提供读服务。

Redis 主从之间数据同步采用异步复制方式，主节点写入成功后，马上返回给客户端，而后异步复制给从节点。

若是数据写入主节点成功，可是还未复制给从节点。此时主节点挂了，从节点马上被提高为主节点。

这种状况下，还未同步的数据就丢失了，其余线程又能够被加锁了。

针对这种状况， Redis 官方提出一种 RedLock 的算法，须要有 N 个Redis 主从节点，解决该问题，详情参考：

redis.io/topics/dist…

这个算法本身实现仍是很复杂的，幸亏 redisson 已经实现的 RedLock，详情参考：redisson redlock

总结

原本这篇文章是想写 Redis 可重入分布式锁的，但是没想到写分布式锁的实现方案就已经写了这么多，再写下去，文章可能就很长，因此拆分红两篇来写。

嘿嘿，这不下星期不用想些什么了，真是个小机灵鬼~

好了，帮你们再次总结一下本文内容。

简单的 Redis 分布式锁的实现方式仍是很简单的，咱们能够直接用 SETNX/DEL 命令实现加解锁。

不过这种实现方式不够健壮，可能存在应用宕机，锁就没法被释放的问题。

因此咱们接着引入如下命令以及 Lua 脚本加强 Redis 分布式锁。

SET lock_name anystring NX EX lock_time
复制代码

最后 Redis 分布锁仍是存在一些缺陷，在这里提出一些解决方案，感兴趣同窗能够本身实现一下。

下篇文章再来将将 Redis 可重入分布式锁~

参考资料

1. 分布式锁的实现之 redis 篇
2. 基于 Redis 的分布式锁

欢迎关注个人公众号：程序通事，得到平常干货推送。若是您对个人专题内容感兴趣，也能够关注个人博客：studyidea.cn