分布式锁（Redis）

一、序言

本文和大家聊聊分布式锁以及常见的解决方案。

二、什么是分布式锁

未命名文件 (4).png
假设一个场景：一个库存服务部署在上面三台机器上，数据库里有 100 件库存，现有 300 个客户同时下单。并且这 300 个客户均摊到上面的三台机器上（即三台机器上分别有 100 个客户）。如果库存服务采取的是传统的进程锁或线程锁，我们会发现三台机器上在检测库存时都能满足（因为每台机器有 100 个客户，刚好满足 100 件库存）。此时会出现只有 100 件库存，却卖出了 300 件的现象（即超卖现象）。
未命名文件 (6).png
为了解决上述在分布式环境中存在的问题，我们需要使用分布式锁。分布式锁是一种在分布式系统中实现线程或进程同步访问共享资源的机制。它的主要目标是在分布式环境下，确保在同一时间只有一个线程或进程可以访问特定的资源。

三、分布式锁方案

分布式锁的实现方式主要有三种：

基于数据库的分布式锁：这种方案主要是在数据库中创建一个表，表中包含方法名等字段，并在方法名字段上创建唯一索引。想要执行某个方法，就使用这个方法名向表中插入数据，成功插入则获取锁，执行完成后删除对应的行数据释放锁。这种方案的缺点包括数据库单点问题、没有锁超时机制、不可重入、非公平锁、非阻塞锁等。
基于 Redis 的分布式锁：Redis 的分布式锁主要是通过 SETNX 和 EXPIRE 命令来实现的。SETNX 可以用来尝试获取锁。EXPIRE 命令用来设置锁的超时时间，防止死锁。此外，还有基于 Redlock 算法的 Redisson 分布式锁。
基于 Zookeeper 的分布式锁：Zookeeper 是一个开源的分布式协调服务，它提供了一种高效且可靠的分布式锁实现机制。Zookeeper 的分布式锁主要是通过临时顺序节点和使用 watch 机制来实现。

四、Redis 分布式锁

4.1 Redis 分布式锁实现方式

Redis 分布式锁的实现通常基于 Redis 的原子性操作（比如 SETNX、EXPIRE、DEL 等），主要思想是通过在Redis 中设置一个特定的键值对来表示锁的状态，当某个节点需要获取锁时，会尝试在 Redis 中设置这个键值对，如果设置成功，则获取到锁，可以执行相应的操作；如果设置失败，则表示锁已经被其他节点持有，当前节点需要等待或重试。

public class RedisLock {private Jedis jedis;private String lockKey;// 构造器public RedisLock(Jedis jedis) {this.jedis = jedis;this.lockKey = "lock";}// 获取锁public boolean tryLock() {// 使用 set key value NX 命令尝试获取锁String result = jedis.set(lockKey, "1", SetParams.setParams().nx());return "OK".equals(result);}// 释放锁public void unLock() {// 释放锁，即删除对应的键jedis.del(lockKey);}
}

4.2 Redis 分布式锁过期

在之前，我们利用 set key value nx 这个互斥命令实现了最基本的分布式锁。但是，现在有一个问题：如果有一个业务在获取锁之后，由于未知原因发生了业务阻塞或者在业务完成之后忘记了释放锁，这将会导致当前业务会永久性的持有该锁。为了解决 Redis 分布式锁无法释放的问题，我们采用给锁设置超时时间：

public class RedisLock {private Jedis jedis;private String lockKey;// 构造器public RedisLock(Jedis jedis) {this.jedis = jedis;this.lockKey = "lock";}// 获取锁public boolean tryLock() {// 使用 set key value NX EX seconds 命令尝试获取锁, 并设置过期时间String result = jedis.set(lockKey, "1", SetParams.setParams().nx().ex(5));return "OK".equals(result);}// 释放锁public void unLock() {// 释放锁，即删除对应的键jedis.del(lockKey);}
}

在上面的代码中，我们利用 set key value nx ex seconds 命令给锁设定了超时时间解决了 Redis 分布式锁被占用而无法释放的问题（设定了超时时间，就算发生了业务阻塞，锁最终也会被释放）。

4.3 Redis 分布式锁误解锁

上面的 Redis 分布式锁引入了超时机制后会带来一个问题。我们先假设一个场景：

业务 A 获取到锁之后发生了业务阻塞，锁被超时释放了。
业务 B 正常获取到锁执行业务。此时，业务 A 恢复执行，并在执行完成后释放掉了锁（此时锁是属于业务 B 的）
业务 C 争抢到锁，但是业务 B 与业务 C 是互斥的此时就会导致并发问题（业务 B 与业务 C 是互斥的，但是同时在执行）。

public class RedisLock {private Jedis jedis;private String lockKey;// 构造器public RedisLock(Jedis jedis, String lockKey) {this.jedis = jedis;// lockKey 不再为固定值this.lockKey = lockKey;}// 获取锁public boolean tryLock() {// 使用 set key value NX EX seconds 命令尝试获取锁, 并设置过期时间String result = jedis.set(lockKey, "1", SetParams.setParams().nx().ex(5));return "OK".equals(result);}// 释放锁public void unLock() {// 释放锁，即删除对应的键jedis.del(lockKey);}
}

为了解决分布式锁误解锁的问题，Redis 分布式锁的 key 不再为一个固定值。业务 A 有自己的 lockKey，业务 B 与业务 C 有相同的 lockKey。此时，业务 A 只能释放自己的锁，业务 B 与业务 C 拥有相同的 lockKey，当业务 B 没有释放锁时，业务 C 是无法获取到锁的，从而保证了业务 B 与业务 C 的互斥。

4.4 Redis 分布式锁续约

在 Redis 分布式锁误解锁的例子中，我们似乎使用不同的 lockKey 解决了误解锁的问题。但是当我们再深入思考一下会发现还有一个问题。我们现假设：

首先，业务 B 获取到锁之后发生了业务阻塞，锁被超时释放了。
然后，业务 C 争抢到锁，由于业务 B 与业务 C 是互斥的此时就会导致并发问题。

每一个业务的执行时间大抵是不尽相同的。在之前的例子中我们使用 jedis.set(lockKey, "1", SetParams.setParams().nx().ex(5))将锁的的释放时间设置成了 5 秒，如果业务能够在 5 秒内执行完成倒是没什么问题，若发生了业务阻塞或业务执行时间大于我们设定的过期时间呢？
针对以上的问题，我们通常采用一种称为 Watch Dog（看门狗）的机制去解决。

public class RedisLock {private final Jedis jedis;private final String lockKey;private final ScheduledExecutorService executorService;// 构造器public RedisLock(Jedis jedis, String lockKey) {this.jedis = jedis;this.lockKey = lockKey;this.executorService = Executors.newSingleThreadScheduledExecutor();}// 尝试获取锁@SneakyThrowspublic boolean tryLock(long leaseTime, TimeUnit timeUnit)  {String result = jedis.set(lockKey, "1", SetParams.setParams().nx().px(timeUnit.toMillis(leaseTime)));if ("OK".equals(result)) {// 获取锁成功时自动启动 watch dogstartWatchdog(leaseTime, timeUnit);return true;}return false;}// 释放锁public void unlock() {jedis.del(lockKey);stopWatchdog(); // 释放锁时停止 watch dog}// 启动 Watchdogpublic void startWatchdog(long leaseTime, TimeUnit timeUnit) {long leaseTimeMillis = timeUnit.toMillis(leaseTime);// 续期检测时间间隔为租约时间的 1/3long checkIntervalMillis = leaseTimeMillis / 3;executorService.scheduleAtFixedRate(() -> {long ttl = jedis.pttl(lockKey);if (ttl > 0) {// 续约锁jedis.pexpire(lockKey, leaseTimeMillis);} else {// 锁过期后停止 watch dogstopWatchdog();}// 周期性执行任务}, checkIntervalMillis, checkIntervalMillis, TimeUnit.MILLISECONDS);}// 停止 Watchdogpublic void stopWatchdog() {executorService.shutdown();}
}

在上面的代码中采用了 Watch Dog 机制周期性的去给锁续期，在业务完成之后，调用 unlock() 方法便可释放锁，并且停止 Watch Dog。

4.5 Redis 分布式锁重试

现在的 Redis 分布式锁已经解决了一部分问题，但是我们假设一个场景：

有三个业务（业务 A，业务 B，业务 C）同时争抢锁，业务 A 首先抢到了锁
业务 A 的执行时间很短，业务 B 与业务 C 此时应该如何处理

现有两种处理方式：

业务 B 与业务 C 直接返回失败信息
业务 B 与业务 C 自动重试争抢锁

在高并发的场景下，第一种方式业务 B 与业务 C 获取到锁的成功率会很小（因为第一次没抢到就返回失败信息了），第二种方式显然会更高（业务 B 与业务 C 会重试获取锁，如果在重试时锁空闲了便能获取到）。为了系统的稳定性和可靠性我们通常会采用第二种方式。

public class RedisLock {private final Jedis jedis;private final String lockKey;private final ScheduledExecutorService executorService;// 构造器public RedisLock(Jedis jedis, String lockKey) {this.jedis = jedis;this.lockKey = lockKey;this.executorService = Executors.newSingleThreadScheduledExecutor();}// 获取锁@SneakyThrowspublic boolean tryLock(long waitTime, long leaseTime, TimeUnit timeUnit)  {long start = System.currentTimeMillis();long end = start + timeUnit.toMillis(waitTime);// while 循环进行锁重试while (System.currentTimeMillis() < end) {String result = jedis.set(lockKey, "1", SetParams.setParams().nx().px(timeUnit.toMillis(leaseTime)));if ("OK".equals(result)) {// 获取锁成功时自动启动 watch dogstartWatchdog(leaseTime, timeUnit);return true;}// 尝试等待一段时间再重试TimeUnit.MILLISECONDS.sleep(100);}return false;}// 释放锁public void unlock() {jedis.del(lockKey);stopWatchdog(); // 释放锁时停止 watch dog}// 启动 Watchdogpublic void startWatchdog(long leaseTime, TimeUnit timeUnit) {long leaseTimeMillis = timeUnit.toMillis(leaseTime);// 续期检测时间间隔为租约时间的 1/3long checkIntervalMillis = leaseTimeMillis / 3;executorService.scheduleAtFixedRate(() -> {long ttl = jedis.pttl(lockKey);if (ttl > 0) {// 续约锁jedis.pexpire(lockKey, leaseTimeMillis);} else {// 锁过期后停止 watch dogstopWatchdog();}}, checkIntervalMillis, checkIntervalMillis, TimeUnit.MILLISECONDS);}// 停止 Watchdogpublic void stopWatchdog() {executorService.shutdown();}
}