Redis缓存数据库进阶——Redis与分布式锁（6）

分布式锁简介

1. 什么是分布式锁

分布式锁是一种在分布式系统环境下，通过多个节点对共享资源进行访问控制的一种同步机制。它的主要目的是防止多个节点同时操作同一份数据，从而避免数据的不一致性。

线程锁：也被称为互斥锁（Mutex），主要用于控制同一进程中的多个线程对共享资源的访问。
进程锁 进程锁是用于控制同一台机器上的多个进程对共享资源的访问。进程锁可以是系统级的，如文件锁，也可以是用户级的，如信号量（Semaphore）。
分布式锁 分布式锁是用于控制分布式系统中的多个节点对共享资源的访问。由于分布式系统中的节点可能位于不同的机器甚至不同的地理位置，因此分布式锁的实现比线程锁和进程锁要复杂得多。分布式锁需要在网络中的多个节点之间进行协调，以保证锁的唯一性和一致性。

2. 分布式锁的特性

分布式锁主要有以下几个特性：

互斥性：在任何时刻，只有一个节点可以持有锁。
不会发生死锁：如果一个节点崩溃，锁可以被其他节点获取。
公平性：如果多个节点同时申请锁，系统应该保证每个节点都有获取锁的机会。
可重入性：同一个节点可以多次获取同一个锁，而不会被阻塞。
高可用：锁服务应该是高可用的，不能因为锁服务的故障而影响整个系统的运行。

分布式锁的基本原理

1. 分布式锁的基本步骤

分布式锁的基本原理可以分为以下几个步骤：

请求锁：当一个实例需要访问共享资源时，它会向分布式锁系统发送一个请求，试图获取一个锁。
锁定资源：分布式锁系统会检查是否有其他实例已经持有这个锁。如果没有，那么这个实例就会获得锁，并且有权访问共享资源。如果有，那么这个实例就必须等待，直到锁被释放。
访问资源：一旦实例获取了锁，它就可以安全地访问共享资源，而不用担心其他实例会同时访问这个资源。
释放锁：当实例完成对共享资源的访问后，它需要通知分布式锁系统释放锁。这样，其他正在等待的实例就可以获取锁，访问共享资源。

SETNX 命令：SETNX（Set if Not Exists）命令用于在 key 不存在时设置值。这是实现分布式锁的关键命令，因为它能确保在同一时间只有一个客户端能够获得锁。
EXPIRE 命令：EXPIRE 命令用于为 key 设置过期时间。这对于避免死锁非常重要，因为即使某个客户端崩溃，锁也会在一定时间后自动释放。
DEL 命令：DEL 命令用于删除 key。在释放锁时，需要使用此命令删除对应的 key。
Lua 脚本：Redis 支持使用 Lua 脚本来执行一系列原子操作。这对于实现安全的分布式锁非常有用，因为它可以确保在释放锁时检查锁的持有者。
RedLock 算法：Redis 官方推荐了一种名为 RedLock 的分布式锁算法。RedLock 是一种基于多个 Redis 实例的分布式锁算法，旨在提供更高的安全性和容错能力。

Redis中分布式锁的实现

Redis分布式锁最简单的实现

SET my_key my_value NX EX 10 # 设置键值对, 超时时间为10s。如果my_key存在，则不进行任何操作

想要实现分布式锁，必须要求 Redis 有「互斥」的能力，我们可以使用 SETNX 命令，这个命令表示SET if Not Exists，即如果 key 不存在，才会设置它的值，否则什么也不做。

两个客户端进程可以执行这个命令，达到互斥，就可以实现一个分布式锁。

客户端 1 申请加锁，加锁成功：

客户端 2 申请加锁，因为它后到达，加锁失败：

此时，加锁成功的客户端，就可以去操作「共享资源」，例如，修改 MySQL 的某一行数据，或者调用一个 API 请求。

操作完成后，还要及时释放锁，给后来者让出操作共享资源的机会。如何释放锁呢？

也很简单，直接使用 DEL 命令删除这个 key 即可，这个逻辑非常简单。

但是，它存在一个很大的问题，当客户端 1 拿到锁后，如果发生下面的场景，就会造成「死锁」：

1、程序处理业务逻辑异常，没及时释放锁

2、进程挂了，没机会释放锁

这时，这个客户端就会一直占用这个锁，而其它客户端就「永远」拿不到这把锁了。怎么解决这个问题呢？

如何避免死锁？

我们很容易想到的方案是，在申请锁时，给这把锁设置一个「租期」。

在 Redis 中实现时，就是给这个 key 设置一个「过期时间」。这里我们假设，操作共享资源的时间不会超过 10s，那么在加锁时，给这个 key 设置 10s 过期即可：

这样一来，无论客户端是否异常，这个锁都可以在 10s 后被「自动释放」，其它客户端依旧可以拿到锁。

但现在还是有问题：

现在的操作，加锁、设置过期是 2 条命令，有没有可能只执行了第一条，第二条却「来不及」执行的情况发生呢？例如：

SETNX 执行成功，执行EXPIRE 时由于网络问题，执行失败
SETNX 执行成功，Redis 异常宕机，EXPIRE 没有机会执行
SETNX 执行成功，客户端异常崩溃，EXPIRE也没有机会执行

总之，这两条命令不能保证是原子操作（一起成功），就有潜在的风险导致过期时间设置失败，依旧发生「死锁」问题。

在 Redis 2.6.12 之后，Redis 扩展了 SET 命令的参数，用这一条命令就可以了：

SET lock 1 EX 10 NX

锁被别人释放怎么办？

上面的命令执行时，每个客户端在释放锁时，都是「无脑」操作，并没有检查这把锁是否还「归自己持有」，所以就会发生释放别人锁的风险，这样的解锁流程，很不「严谨」！如何解决这个问题呢？

解决办法是：客户端在加锁时，设置一个只有自己知道的「唯一标识」进去。

例如，可以是自己的线程 ID，也可以是一个 UUID（随机且唯一），这里我们以UUID 举例：

SET lock $uuid EX 20 NX

之后，在释放锁时，要先判断这把锁是否还归自己持有，伪代码可以这么写：

if redis.get("lock") == $uuid:
redis.del("lock")

这里释放锁使用的是 GET + DEL 两条命令，这时，又会遇到我们前面讲的原子性问题了。这里可以使用lua脚本来解决。

安全释放锁的 Lua 脚本如下：

if redis.call("GET",KEYS[1]) == ARGV[1]
then
return redis.call("DEL",KEYS[1])
else
return 0
end

好了，这样一路优化，整个的加锁、解锁的流程就更「严谨」了。

这里我们先小结一下，基于 Redis 实现的分布式锁，一个严谨的的流程如下：

1、加锁

SET lock_key $unique_id EX $expire_time NX

2、操作共享资源

3、释放锁：Lua 脚本，先 GET 判断锁是否归属自己，再DEL 释放锁

Java代码实现分布式锁

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.params.SetParams;import java.util.Arrays;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;/*** 分布式锁的实现*/
@Component
public class RedisDistLock implements Lock {private final static int LOCK_TIME = 5*1000;private final static String RS_DISTLOCK_NS = "tdln:";/*if redis.call('get',KEYS[1])==ARGV[1] thenreturn redis.call('del', KEYS[1])else return 0 end*/private final static String RELEASE_LOCK_LUA ="if redis.call('get',KEYS[1])==ARGV[1] then\n" +"        return redis.call('del', KEYS[1])\n" +"    else return 0 end";/*保存每个线程的独有的ID值*/private ThreadLocal<String> lockerId = new ThreadLocal<>();/*解决锁的重入*/private Thread ownerThread;private String lockName = "lock";@Autowiredprivate JedisPool jedisPool;public String getLockName() {return lockName;}public void setLockName(String lockName) {this.lockName = lockName;}public Thread getOwnerThread() {return ownerThread;}public void setOwnerThread(Thread ownerThread) {this.ownerThread = ownerThread;}@Overridepublic void lock() {while(!tryLock()){try {Thread.sleep(100);} catch (InterruptedException e) {e.printStackTrace();}}}@Overridepublic void lockInterruptibly() throws InterruptedException {throw new UnsupportedOperationException("不支持可中断获取锁！");}@Overridepublic boolean tryLock() {Thread t = Thread.currentThread();if(ownerThread==t){/*说明本线程持有锁*/return true;}else if(ownerThread!=null){/*本进程里有其他线程持有分布式锁*/return false;}Jedis jedis = null;try {String id = UUID.randomUUID().toString();SetParams params = new SetParams();params.px(LOCK_TIME);params.nx();synchronized (this){/*线程们，本地抢锁*/if((ownerThread==null)&&"OK".equals(jedis.set(RS_DISTLOCK_NS+lockName,id,params))){lockerId.set(id);setOwnerThread(t);return true;}else{return false;}}} catch (Exception e) {throw new RuntimeException("分布式锁尝试加锁失败！");} finally {jedis.close();}}@Overridepublic boolean tryLock(long time, TimeUnit unit) throws InterruptedException {throw new UnsupportedOperationException("不支持等待尝试获取锁！");}@Overridepublic void unlock() {if(ownerThread!=Thread.currentThread()) {throw new RuntimeException("试图释放无所有权的锁！");}Jedis jedis = null;try {jedis = jedisPool.getResource();Long result = (Long)jedis.eval(RELEASE_LOCK_LUA,Arrays.asList(RS_DISTLOCK_NS+lockName),Arrays.asList(lockerId.get()));if(result.longValue()!=0L){System.out.println("Redis上的锁已释放！");}else{System.out.println("Redis上的锁释放失败！");}} catch (Exception e) {throw new RuntimeException("释放锁失败！",e);} finally {if(jedis!=null) jedis.close();lockerId.remove();setOwnerThread(null);System.out.println("本地锁所有权已释放！");}}@Overridepublic Condition newCondition() {throw new UnsupportedOperationException("不支持等待通知操作！");}}

锁过期时间不好评估怎么办？

看上面这张图，假入key的失效时间是10s，但是客户端C在拿到分布式锁之后，然后业务逻辑执行超过10s，那么问题来了，在客户端C释放锁之前，其实这把锁已经失效了，那么客户端A和客户端B都可以去拿锁，这样就已经失去了分布式锁的功能了！！！

比较简单的妥协方案是，尽量「冗余」过期时间，降低锁提前过期的概率，但是这个并不能完美解决问题，那怎么办呢？

分布式锁加入看门狗

加锁时，先设置一个过期时间，然后我们开启一个「守护线程」，定时去检测这个锁的失效时间，如果锁快要过期了，操作共享资源还未完成，那么就自动对锁进行「续期」，重新设置过期时间。

这个守护线程我们一般也把它叫做「看门狗」线程。

为什么要使用守护线程：

Redisson中的分布式锁

Redisson把这些工作都封装好了

<dependency><groupId>org.redisson</groupId><artifactId>redisson</artifactId><version>3.12.3</version></dependency>import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class MyRedissonConfig {/*** 所有对Redisson的使用都是通过RedissonClient*/@Bean(destroyMethod="shutdown")public RedissonClient redisson(){//1、创建配置Config config = new Config();config.useSingleServer().setAddress("redis://127.0.0.1:6379");//2、根据Config创建出RedissonClient实例RedissonClient redisson = Redisson.create(config);return redisson;}
}import com.msb.redis.lock.rdl.RedisDistLockWithDog;
import org.junit.jupiter.api.Test;
import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;@SpringBootTest
public class TestRedissionLock {private int count = 0;@Autowiredprivate RedissonClient redisson;@Testpublic void testLockWithDog() throws InterruptedException {int clientCount =3;RLock lock = redisson.getLock("RD-lock");CountDownLatch countDownLatch = new CountDownLatch(clientCount);ExecutorService executorService = Executors.newFixedThreadPool(clientCount);for (int i = 0;i<clientCount;i++){executorService.execute(() -> {try {lock.lock(10, TimeUnit.SECONDS);System.out.println(Thread.currentThread().getName()+"准备进行累加。");Thread.sleep(2000);count++;} catch (InterruptedException e) {e.printStackTrace();} finally {lock.unlock();}countDownLatch.countDown();});}countDownLatch.await();System.out.println(count);}
}

GitHub - redisson/redisson: Redisson - Easy Redis Java client and Real-Time Data Platform. Valkey compatible. Sync/Async/RxJava/Reactive API. Over 50 Redis based Java objects and services: Set, Multimap, SortedSet, Map, List, Queue, Deque, Semaphore, Lock, AtomicLong, Map Reduce, Bloom filter, Spring Cache, Tomcat, Scheduler, JCache API, Hibernate, RPC, local cache ...

Redisson: Easy Redis Java client and Real-Time Data Platform

锁过期时间不好评估怎么办？

集群下的锁还安全么？

基于 Redis 的实现分布式锁，前面遇到的问题，以及对应的解决方案：

1、死锁：设置过期时间

2、过期时间评估不好，锁提前过期：守护线程，自动续期

3、锁被别人释放：锁写入唯一标识，释放锁先检查标识，再释放

之前分析的场景都是，锁在「单个」Redis实例中可能产生的问题，并没有涉及到 Redis 的部署架构细节。

而我们在使用 Redis 时，一般会采用主从集群 +哨兵的模式部署，这样做的好处在于，当主库异常宕机时，哨兵可以实现「故障自动切换」，把从库提升为主库，继续提供服务，以此保证可用性。

但是因为主从复制是异步的，那么就不可避免会发生的锁数据丢失问题（加了锁却没来得及同步过来）。从库被哨兵提升为新主库，这个锁在新的主库上，丢失了！

Redlock真的安全吗？

Redis 作者提出的 Redlock方案，是如何解决主从切换后，锁失效问题的。

Redlock 的方案基于一个前提：

不再需要部署从库和哨兵实例，只部署主库；但主库要部署多个，官方推荐至少 5 个实例。

注意：不是部署 Redis Cluster，就是部署 5 个简单的 Redis 实例。它们之间没有任何关系，都是一个个孤立的实例。

做完之后，我们看官网代码怎么去用的：

8. 分布式锁和同步器 · redisson/redisson Wiki · GitHub

8.4. 红锁（RedLock）

基于Redis的Redisson红锁 RedissonRedLock对象实现了Redlock介绍的加锁算法。该对象也可以用来将多个 RLock对象关联为一个红锁，每个 RLock对象实例可以来自于不同的Redisson实例。

RLock lock1 = redissonInstance1.getLock("lock1");
RLock lock2 = redissonInstance2.getLock("lock2");
RLock lock3 = redissonInstance3.getLock("lock3");RedissonRedLock lock = new RedissonRedLock(lock1, lock2, lock3);
// 同时加锁：lock1 lock2 lock3
// 红锁在大部分节点上加锁成功就算成功。
lock.lock();
...
lock.unlock();

大家都知道，如果负责储存某些分布式锁的某些Redis节点宕机以后，而且这些锁正好处于锁住的状态时，这些锁会出现锁死的状态。为了避免这种情况的发生，Redisson内部提供了一个监控锁的看门狗，它的作用是在Redisson实例被关闭前，不断的延长锁的有效期。默认情况下，看门狗的检查锁的超时时间是30秒钟，也可以通过修改Config.lockWatchdogTimeout来另行指定。

另外Redisson还通过加锁的方法提供了 leaseTime的参数来指定加锁的时间。超过这个时间后锁便自动解开了。

RedissonRedLock lock = new RedissonRedLock(lock1, lock2, lock3);
// 给lock1，lock2，lock3加锁，如果没有手动解开的话，10秒钟后将会自动解开
lock.lock(10, TimeUnit.SECONDS);// 为加锁等待100秒时间，并在加锁成功10秒钟后自动解开
boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);
...
lock.unlock();

Redlock实现整体流程

1、客户端先获取「当前时间戳T1」

2、客户端依次向这 5 个 Redis 实例发起加锁请求

3、如果客户端从 >=3 个（大多数）以上Redis 实例加锁成功，则再次获取「当前时间戳T2」，如果 T2 - T1 < 锁的过期时间，此时，认为客户端加锁成功，否则认为加锁失败。

4、加锁成功，去操作共享资源

5、加锁失败/释放锁，向「全部节点」发起释放锁请求。

所以总的来说：客户端在多个 Redis 实例上申请加锁；必须保证大多数节点加锁成功；大多数节点加锁的总耗时，要小于锁设置的过期时间；释放锁，要向全部节点发起释放锁请求。

我们来看 Redlock 为什么要这么做？

为什么要在多个实例上加锁？

本质上是为了「容错」，部分实例异常宕机，剩余的实例加锁成功，整个锁服务依旧可用。

为什么大多数加锁成功，才算成功？

多个 Redis 实例一起来用，其实就组成了一个「分布式系统」。在分布式系统中，总会出现「异常节点」，所以，在谈论分布式系统问题时，需要考虑异常节点达到多少个，也依旧不会影响整个系统的「正确性」。

这是一个分布式系统「容错」问题，这个问题的结论是：如果只存在「故障」节点，只要大多数节点正常，那么整个系统依旧是可以提供正确服务的。

为什么步骤 3 加锁成功后，还要计算加锁的累计耗时？

因为操作的是多个节点，所以耗时肯定会比操作单个实例耗时更久，而且，因为是网络请求，网络情况是复杂的，有可能存在延迟、丢包、超时等情况发生，网络请求越多，异常发生的概率就越大。

所以，即使大多数节点加锁成功，但如果加锁的累计耗时已经「超过」了锁的过期时间，那此时有些实例上的锁可能已经失效了，这个锁就没有意义了。

为什么释放锁，要操作所有节点？

在某一个 Redis 节点加锁时，可能因为「网络原因」导致加锁失败。

例如，客户端在一个 Redis 实例上加锁成功，但在读取响应结果时，网络问题导致读取失败，那这把锁其实已经在 Redis 上加锁成功了。

所以，释放锁时，不管之前有没有加锁成功，需要释放「所有节点」的锁，以保证清理节点上「残留」的锁。

好了，明白了 Redlock 的流程和相关问题，看似Redlock 确实解决了 Redis 节点异常宕机锁失效的问题，保证了锁的「安全性」。

但事实真的如此吗？

RedLock的是是非非

一个分布式系统，更像一个复杂的「野兽」，存在着你想不到的各种异常情况。

这些异常场景主要包括三大块，这也是分布式系统会遇到的三座大山：NPC。

N：Network Delay，网络延迟

P：Process Pause，进程暂停（GC）

C：Clock Drift，时钟漂移

比如一个进程暂停（GC）的例子

1）客户端 1 请求锁定节点 A、B、C、D、E

2）客户端 1 的拿到锁后，进入 GC（时间比较久）

3）所有 Redis 节点上的锁都过期了

4）客户端 2 获取到了 A、B、C、D、E 上的锁

5）客户端 1 GC 结束，认为成功获取锁

6）客户端 2 也认为获取到了锁，发生「冲突」

GC 和网络延迟问题：这两点可以在红锁实现流程的第3步来解决这个问题。

但是最核心的还是时钟漂移，因为时钟漂移，就有可能导致第3步的判断本身就是一个BUG，所以当多个 Redis 节点「时钟」发生问题时，也会导致 Redlock 锁失效。

RedLock总结

Redlock 只有建立在「时钟正确」的前提下，才能正常工作，如果你可以保证这个前提，那么可以拿来使用。

但是时钟偏移在现实中是存在的：

第一，从硬件角度来说，时钟发生偏移是时有发生，无法避免。例如，CPU 温度、机器负载、芯片材料都是有可能导致时钟发生偏移的。

第二，人为错误也是很难完全避免的。

所以，Redlock尽量不用它，而且它的性能不如单机版 Redis，部署成本也高，优先考虑使用主从+ 哨兵的模式实现分布式锁（只会有很小的记录发生主从切换时的锁丢失问题）。

分布式锁的常见问题和解决方案

1. 死锁问题

问题： 当一个客户端获取了锁，但由于某些原因（如程序崩溃、异常等）无法释放锁时，会导致其他客户端永远无法获取锁。
解决方案： 设置锁的过期时间。当锁的持有者未能在过期时间内执行完毕并释放锁时，锁将自动过期，从而允许其他客户端获取锁。

2. 锁续命问题

问题： 如果一个操作需要的时间可能超过锁的过期时间，那么在操作执行过程中锁过期会导致其他客户端获取到锁，从而产生并发问题。 解决方案： 使用锁续命机制。在锁持有者执行操作期间，可以定期检查锁是否即将过期，并在适当的时候对锁进行续命，即重新设置锁的过期时间。

3. 锁释放问题

问题： 为确保数据的一致性，只有锁的持有者才能释放锁。但在实际应用中，可能会出现误解锁的情况。
解决方案： 在设置锁时，为锁关联一个唯一的值（如UUID）。在释放锁时，先检查锁的值是否与当前客户端的值匹配，如果匹配则释放锁，否则不做任何操作。注意，锁持有人的判断和锁的释放应该在一个原子操作内完成。