共计 15541 个字符，预计需要花费 39 分钟才能阅读完成。
这篇文章主要介绍“如何理解分布式系统下基于 Redis 的分布式锁”，在日常操作中，相信很多人在如何理解分布式系统下基于 Redis 的分布式锁问题上存在疑惑，丸趣 TV 小编查阅了各式资料，整理出简单好用的操作方法，希望对大家解答”如何理解分布式系统下基于 Redis 的分布式锁”的疑惑有所帮助！接下来，请跟着丸趣 TV 小编一起来学习吧！
新接手的项目，偶尔会出现账不平的问题。之前的技术老大临走时给的解释是：排查了，没找到原因，之后太忙就没再解决，可能是框架的原因……
既然项目交付到手中，这样的问题是必须要解决的。梳理了所有账务处理逻辑，最终找到了原因：数据库并发操作热点账户导致。就这这个问题，来聊一聊分布式系统下基于 Redis 的分布式锁。顺便也分解一下问题形成原因及解决方案。
原因分析
系统并发量并不高，存在热点账户，但也不至于那么严重。问题的根源在于系统架构设计，人为的制造了并发。场景是这样的：商户批量导入一批数据，系统会进行前置处理，并对账户余额进行增减。
此时，另外一个定时任务，也会对账户进行扫描更新。而且对同一账户的操作分布到各个系统当中，热点账户也就出现了。
针对此问题的解决方案，从架构层面可以考虑将账务系统进行抽离，集中在一个系统中进行处理，所有的数据库事务及执行顺序由账务系统来统筹处理。从技术方面来讲，则可以通过锁机制来对热点账户进行加锁。
本篇文章就针对热点账户基于分布式锁的实现方式进行详细的讲解。
锁的分析
在 Java 的多线程环境下，通常有几类锁可以使用：
JVM 内存模型级别的锁，常用的有：synchronized、Lock 等；
数据库锁，比如乐观锁，悲观锁等；
分布式锁；
JVM 内存级别的锁，可以保证单体服务下线程的安全性，比如多个线程访问 / 修改一个全局变量。但当系统进行集群部署时，JVM 级别的本地锁就无能为力了。
悲观锁与乐观锁
像上述案例中，热点账户就属于分布式系统中的共享资源，我们通常会采用数据库锁或分布式锁来进行解决。
数据库锁，又分为乐观锁和悲观锁。
悲观锁是基于数据库（Mysql 的 InnoDB）提供的排他锁来实现的。在进行事务操作时，通过 select … for update 语句，MySQL 会对查询结果集中每行数据都添加排他锁，其他线程对该记录的更新与删除操作都会阻塞。从而达到共享资源的顺序执行（修改）；
乐观锁是相对悲观锁而言的，乐观锁假设数据一般情况不会造成冲突，所以在数据进行提交更新的时候，才会正式对数据的冲突与否进行检测。如果冲突则返回给用户异常信息，让用户决定如何去做。乐观锁适用于读多写少的场景，这样可以提高程序的吞吐量。在乐观锁实现时通常会基于记录状态或添加 version 版本来进行实现。
悲观锁失效场景
项目中使用了悲观锁，但悲观锁却失效了。这也是使用悲观锁时，常见的误区，下面来分析一下。
正常使用悲观锁的流程：
通过 select … for update 锁定记录；
计算新余额，修改金额并存储；
执行完成释放锁；
经常犯错的处理流程：
查询账户余额，计算新余额；
通过 select … for update 锁定记录；
修改金额并存储；
执行完成释放锁；
错误的流程中，比如 A 和 B 服务查询到的余额都是 100，A 扣减 50，B 扣减 40，然后 A 锁定记录，更新数据库为 50；A 释放锁之后，B 锁定记录，更新数据库为 60。显然，后者把前者的更新给覆盖掉了。解决的方案就是扩大锁的范围，将锁提前到计算新余额之前。
通常悲观锁对数据库的压力是非常大的，在实践中通常会根据场景使用乐观锁或分布式锁等方式来实现。
下面进入正题，讲讲基于 Redis 的分布式锁实现。
Redis 分布式锁实战演习
这里以 Spring Boot、Redis、Lua 脚本为例来演示分布式锁的实现。为了简化处理，示例中 Redis 既承担了分布式锁的功能，也承担了数据库的功能。
场景构建
集群环境下，对同一个账户的金额进行操作，基本步骤：
从数据库读取用户金额；
程序修改金额；
再将最新金额存储到数据库；
下面从最初不加锁，不同步处理，逐步推演出最终的分布式锁。
基础集成及类构建
准备一个不加锁处理的基础业务环境。
首先在 Spring Boot 项目中引入相关依赖：
dependency
groupId org.springframework.boot /groupId
artifactId spring-boot-starter-data-redis /artifactId
/dependency
dependency
groupId org.springframework.boot /groupId
artifactId spring-boot-starter-web /artifactId
/dependency账户对应实体类 UserAccount：
public class UserAccount {
// 用户 ID
private String userId;
// 账户内金额
private int amount;
// 添加账户金额
public void addAmount(int amount) {
this.amount = this.amount + amount;
}
// 省略构造方法和 getter/setter
}创建一个线程实现类 AccountOperationThread：
public class AccountOperationThread implements Runnable { private final static Logger logger = LoggerFactory.getLogger(AccountOperationThread.class);
private static final Long RELEASE_SUCCESS = 1L;
private String userId;
private RedisTemplate Object, Object redisTemplate;
public AccountOperationThread(String userId, RedisTemplate Object, Object redisTemplate) {
this.userId = userId;
this.redisTemplate = redisTemplate;
}
@Override
public void run() { noLock();
}
/**
* 不加锁
*/
private void noLock() {
try { Random random = new Random();
// 模拟线程进行业务处理
TimeUnit.MILLISECONDS.sleep(random.nextInt(100) + 1);
} catch (InterruptedException e) { e.printStackTrace();
}
// 模拟数据库中获取用户账号
UserAccount userAccount = (UserAccount) redisTemplate.opsForValue().get(userId);
// 金额 +1
userAccount.addAmount(1);
logger.info(Thread.currentThread().getName() + : user id : + userId + amount : + userAccount.getAmount());
// 模拟存回数据库
redisTemplate.opsForValue().set(userId, userAccount);
}
}其中 RedisTemplate 的实例化交给了 Spring Boot：
@Configuration
public class RedisConfig {
@Bean
public RedisTemplate Object, Object redisTemplate(RedisConnectionFactory redisConnectionFactory) { RedisTemplate Object, Object redisTemplate = new RedisTemplate ();
redisTemplate.setConnectionFactory(redisConnectionFactory);
Jackson2JsonRedisSerializer Object jackson2JsonRedisSerializer =
new Jackson2JsonRedisSerializer (Object.class);
ObjectMapper objectMapper = new ObjectMapper();
objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
jackson2JsonRedisSerializer.setObjectMapper(objectMapper);
// 设置 value 的序列化规则和 key 的序列化规则
redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);
redisTemplate.setKeySerializer(new StringRedisSerializer());
redisTemplate.afterPropertiesSet();
return redisTemplate;
}
}最后，再准备一个 TestController 来进行触发多线程的运行：
@RestController
public class TestController { private final static Logger logger = LoggerFactory.getLogger(TestController.class);
private static ExecutorService executorService = Executors.newFixedThreadPool(10);
@Autowired
private RedisTemplate Object, Object redisTemplate;
@GetMapping(/test)
public String test() throws InterruptedException {
// 初始化用户 user_001 到 Redis，账户金额为 0
redisTemplate.opsForValue().set( user_001 , new UserAccount( user_001 , 0));
// 开启 10 个线程进行同步测试，每个线程为账户增加 1 元
for (int i = 0; i 10; i++) { logger.info( 创建线程 i = + i);
executorService.execute(new AccountOperationThread( user_001 , redisTemplate));
}
// 主线程休眠 1 秒等待线程跑完
TimeUnit.MILLISECONDS.sleep(1000);
// 查询 Redis 中的 user_001 账户
UserAccount userAccount = (UserAccount) redisTemplate.opsForValue().get( user_001
logger.info(user id : + userAccount.getUserId() + amount : + userAccount.getAmount());
return success
}
}执行上述程序，正常来说 10 个线程，每个线程加 1，结果应该是 10。但多执行几次，会发现，结果变化很大，基本上都要比 10 小。
[pool-1-thread-5] c.s.redis.thread.AccountOperationThread : pool-1-thread-5 : user id : user_001 amount : 1
[pool-1-thread-4] c.s.redis.thread.AccountOperationThread : pool-1-thread-4 : user id : user_001 amount : 1
[pool-1-thread-3] c.s.redis.thread.AccountOperationThread : pool-1-thread-3 : user id : user_001 amount : 1
[pool-1-thread-1] c.s.redis.thread.AccountOperationThread : pool-1-thread-1 : user id : user_001 amount : 1
[pool-1-thread-1] c.s.redis.thread.AccountOperationThread : pool-1-thread-1 : user id : user_001 amount : 2
[pool-1-thread-2] c.s.redis.thread.AccountOperationThread : pool-1-thread-2 : user id : user_001 amount : 2
[pool-1-thread-5] c.s.redis.thread.AccountOperationThread : pool-1-thread-5 : user id : user_001 amount : 2
[pool-1-thread-4] c.s.redis.thread.AccountOperationThread : pool-1-thread-4 : user id : user_001 amount : 3
[pool-1-thread-1] c.s.redis.thread.AccountOperationThread : pool-1-thread-1 : user id : user_001 amount : 4
[pool-1-thread-3] c.s.redis.thread.AccountOperationThread : pool-1-thread-3 : user id : user_001 amount : 5
[nio-8080-exec-1] c.s.redis.controller.TestController : user id : user_001 amount : 5以上述日志为例，前四个线程都将值改为 1，也就是后面三个线程都将前面的修改进行了覆盖，导致最终结果不是 10，只有 5。这显然是有问题的。
Redis 同步锁实现
针对上面的情况，在同一个 JVM 当中，我们可以通过线程加锁来完成。但在分布式环境下，JVM 级别的锁是没办法实现的，这里可以采用 Redis 同步锁实现。
基本思路：第一个线程进入时，在 Redis 中进记录，当后续线程过来请求时，判断 Redis 是否存在该记录，如果存在则说明处于锁定状态，进行等待或返回。如果不存在，则进行后续业务处理。
/**
* 1. 抢占资源时判断是否被锁。 * 2. 如未锁则抢占成功且加锁，否则等待锁释放。 * 3. 业务完成后释放锁, 让给其它线程。 * p
* 该方案并未解决同步问题，原因：线程获得锁和加锁的过程，并非原子性操作，可能会导致线程 A 获得锁，还未加锁时，线程 B 也获得了锁。 */
private void redisLock() { Random random = new Random();
try { TimeUnit.MILLISECONDS.sleep(random.nextInt(1000) + 1);
} catch (InterruptedException e) { e.printStackTrace();
}
while (true) { Object lock = redisTemplate.opsForValue().get(userId + :syn
if (lock == null) {
// 获得锁 - 加锁 - 跳出循环
logger.info(Thread.currentThread().getName() + : 获得锁
redisTemplate.opsForValue().set(userId + :syn , lock
break;
}
try {
// 等待 500 毫秒重试获得锁
TimeUnit.MILLISECONDS.sleep(500);
} catch (InterruptedException e) { e.printStackTrace();
}
}
try {
// 模拟数据库中获取用户账号
UserAccount userAccount = (UserAccount) redisTemplate.opsForValue().get(userId);
if (userAccount != null) {
// 设置金额
userAccount.addAmount(1);
logger.info(Thread.currentThread().getName() + : user id : + userId + amount : + userAccount.getAmount());
// 模拟存回数据库
redisTemplate.opsForValue().set(userId, userAccount);
}
} finally {
// 释放锁
redisTemplate.delete(userId + :syn
logger.info(Thread.currentThread().getName() + : 释放锁
}
}在 while 代码块中，先判断对应用户 ID 是否在 Redis 中存在，如果不存在，则进行 set 加锁，如果存在，则跳出循环继续等待。
上述代码，看起来实现了加锁的功能，但当执行程序时，会发现与未加锁一样，依旧存在并发问题。原因是：获取锁和加锁的操作并不是原子的。比如两个线程发现 lock 都是 null，都进行了加锁，此时并发问题依旧存在。
Redis 原子性同步锁
针对上述问题，可将获取锁和加锁的过程原子化处理。基于 spring-boot-data-redis 提供的原子化 API 可以实现：
// 该方法使用了 redis 的指令：SETNX key value
// 1.key 不存在，设置成功返回 value，setIfAbsent 返回 true；// 2.key 存在，则设置失败返回 null，setIfAbsent 返回 false；// 3. 原子性操作；Boolean setIfAbsent(K var1, V var2);上述方法的原子化操作是对 Redis 的 setnx 命令的封装，在 Redis 中 setnx 的使用如下实例：
redis SETNX mykey Hello
(integer) 1
redis SETNX mykey World
(integer) 0
redis GET mykey
Hello第一次，设置 mykey 时，并不存在，则返回 1，表示设置成功；第二次设置 mykey 时，已经存在，则返回 0，表示设置失败。再次查询 mykey 对应的值，会发现依旧是第一次设置的值。也就是说 redis 的 setnx 保证了唯一的 key 只能被一个服务设置成功。
理解了上述 API 及底层原理，来看看线程中的实现方法代码如下：
/**
* 1. 原子操作加锁
* 2. 竞争线程循环重试获得锁
* 3. 业务完成释放锁
*/
private void atomicityRedisLock() {
//Spring data redis 支持的原子性操作
while (!redisTemplate.opsForValue().setIfAbsent(userId + :syn , lock)) {
try {
// 等待 100 毫秒重试获得锁
TimeUnit.MILLISECONDS.sleep(100);
} catch (InterruptedException e) { e.printStackTrace();
}
}
logger.info(Thread.currentThread().getName() + : 获得锁
try {
// 模拟数据库中获取用户账号
UserAccount userAccount = (UserAccount) redisTemplate.opsForValue().get(userId);
if (userAccount != null) {
// 设置金额
userAccount.addAmount(1);
logger.info(Thread.currentThread().getName() + : user id : + userId + amount : + userAccount.getAmount());
// 模拟存回数据库
redisTemplate.opsForValue().set(userId, userAccount);
}
} finally {
// 释放锁
redisTemplate.delete(userId + :syn
logger.info(Thread.currentThread().getName() + : 释放锁
}
}再次执行代码，会发现结果正确了，也就是说可以成功的对分布式线程进行了加锁。
Redis 分布式锁的死锁
虽然上述代码执行结果没问题，但如果应用异常宕机，没来得及执行 finally 中释放锁的方法，那么其他线程则永远无法获得这个锁。
此时可采用 setIfAbsent 的重载方法：
Boolean setIfAbsent(K var1, V var2, long var3, TimeUnit var5);基于该方法，可以设置锁的过期时间。这样即便获得锁的线程宕机，在 Redis 中数据过期之后，其他线程可正常获得该锁。
示例代码如下：
private void atomicityAndExRedisLock() {
try {
//Spring data redis 支持的原子性操作, 并设置 5 秒过期时间
while (!redisTemplate.opsForValue().setIfAbsent(userId + :syn ,
System.currentTimeMillis() + 5000, 5000, TimeUnit.MILLISECONDS)) {
// 等待 100 毫秒重试获得锁
logger.info(Thread.currentThread().getName() + : 尝试循环获取锁
TimeUnit.MILLISECONDS.sleep(1000);
}
logger.info(Thread.currentThread().getName() + : 获得锁 --------
// 应用在这里宕机，进程退出，无法执行 finally;
Thread.currentThread().interrupt();
// 业务逻辑...
} catch (InterruptedException e) { e.printStackTrace();
} finally {
// 释放锁
if (!Thread.currentThread().isInterrupted()) {
redisTemplate.delete(userId + :syn
logger.info(Thread.currentThread().getName() + : 释放锁
}
}
}业务超时及守护线程
上面添加了 Redis 所的超时时间，看似解决了问题，但又引入了新的问题。
比如，正常情况下线程 A 在 5 秒内可正常处理完业务，但偶发会出现超过 5 秒的情况。如果将超时时间设置为 5 秒，线程 A 获得了锁，但业务逻辑处理需要 6 秒。此时，线程 A 还在正常业务逻辑，线程 B 已经获得了锁。当线程 A 处理完时，有可能将线程 B 的锁给释放掉。
在上述场景中有两个问题点：
第一，线程 A 和线程 B 可能会同时在执行，存在并发问题。
第二，线程 A 可能会把线程 B 的锁给释放掉，导致一系列的恶性循环。
当然，可以通过在 Redis 中设置 value 值来判断锁是属于线程 A 还是线程 B。但仔细分析会发现，这个问题的本质是因为线程 A 执行业务逻辑耗时超出了锁超时的时间。
那么就有两个解决方案了：
第一，将超时时间设置的足够长，确保业务代码能够在锁释放之前执行完成；
第二，为锁添加守护线程，为将要过期释放但未释放的锁增加时间；
第一种方式需要全行大多数情况下业务逻辑的耗时，进行超时时间的设定。
第二种方式，可通过如下守护线程的方式来动态增加锁超时时间。
public class DaemonThread implements Runnable { private final static Logger logger = LoggerFactory.getLogger(DaemonThread.class);
// 是否需要守护 主线程关闭则结束守护线程
private volatile boolean daemon = true;
// 守护锁
private String lockKey;
private RedisTemplate Object, Object redisTemplate;
public DaemonThread(String lockKey, RedisTemplate Object, Object redisTemplate) {
this.lockKey = lockKey;
this.redisTemplate = redisTemplate;
}
@Override
public void run() {
try { while (daemon) { long time = redisTemplate.getExpire(lockKey, TimeUnit.MILLISECONDS);
// 剩余有效期小于 1 秒则续命
if (time 1000) { logger.info( 守护进程: + Thread.currentThread().getName() + 延长锁时间 5000 毫秒
redisTemplate.expire(lockKey, 5000, TimeUnit.MILLISECONDS);
}
TimeUnit.MILLISECONDS.sleep(300);
}
logger.info( 守护进程: + Thread.currentThread().getName() + 关闭
} catch (InterruptedException e) { e.printStackTrace();
}
}
// 主线程主动调用结束
public void stop() {
daemon = false;
}
}上述线程每隔 300 毫秒获取一下 Redis 中锁的超时时间，如果小于 1 秒，则延长 5 秒。当主线程调用关闭时，守护线程也随之关闭。
主线程中相关代码实现：
private void deamonRedisLock() {
// 守护线程
DaemonThread daemonThread = null;
//Spring data redis 支持的原子性操作, 并设置 5 秒过期时间
String uuid = UUID.randomUUID().toString();
String value = Thread.currentThread().getId() + : + uuid;
try { while (!redisTemplate.opsForValue().setIfAbsent(userId + :syn , value, 5000, TimeUnit.MILLISECONDS)) {
// 等待 100 毫秒重试获得锁
logger.info(Thread.currentThread().getName() + : 尝试循环获取锁
TimeUnit.MILLISECONDS.sleep(1000);
}
logger.info(Thread.currentThread().getName() + : 获得锁 ----
// 开启守护线程
daemonThread = new DaemonThread(userId + :syn , redisTemplate);
Thread thread = new Thread(daemonThread);
thread.start();
// 业务逻辑执行 10 秒...
TimeUnit.MILLISECONDS.sleep(10000);
} catch (InterruptedException e) { e.printStackTrace();
} finally {
// 释放锁 这里也需要原子操作, 今后通过 Redis + Lua 讲
String result = (String) redisTemplate.opsForValue().get(userId + :syn
if (value.equals(result)) {
redisTemplate.delete(userId + :syn
logger.info(Thread.currentThread().getName() + : 释放锁 -----
}
// 关闭守护线程
if (daemonThread != null) { daemonThread.stop();
}
}
}其中在获得锁之后，开启守护线程，在 finally 中将守护线程关闭。
基于 Lua 脚本的实现
在上述逻辑中，我们是基于 spring-boot-data-redis 提供的原子化操作来保证锁判断和执行的原子化的。在非 Spring Boot 项目中，则可以基于 Lua 脚本来实现。
首先定义加锁和解锁的 Lua 脚本及对应的 DefaultRedisScript 对象，在 RedisConfig 配置类中添加如下实例化代码：
@Configuration
public class RedisConfig {
//lock script
private static final String LOCK_SCRIPT = if redis.call(setnx ,KEYS[1],ARGV[1]) == 1 +
then redis.call(expire ,KEYS[1],ARGV[2]) +
return 1 +
else return 0 end
private static final String UNLOCK_SCRIPT = if redis.call(get , KEYS[1]) == ARGV[1] then return redis.call +
(del , KEYS[1]) else return 0 end
// ... 省略部分代码
@Bean
public DefaultRedisScript Boolean lockRedisScript() { DefaultRedisScript Boolean defaultRedisScript = new DefaultRedisScript ();
defaultRedisScript.setResultType(Boolean.class);
defaultRedisScript.setScriptText(LOCK_SCRIPT);
return defaultRedisScript;
}
@Bean
public DefaultRedisScript Long unlockRedisScript() { DefaultRedisScript Long defaultRedisScript = new DefaultRedisScript ();
defaultRedisScript.setResultType(Long.class);
defaultRedisScript.setScriptText(UNLOCK_SCRIPT);
return defaultRedisScript;
}
}再通过在 AccountOperationThread 类中新建构造方法，将上述两个对象传入类中（省略此部分演示）。然后，就可以基于 RedisTemplate 来调用了，改造之后的代码实现如下：
private void deamonRedisLockWithLua() {
// 守护线程
DaemonThread daemonThread = null;
//Spring data redis 支持的原子性操作, 并设置 5 秒过期时间
String uuid = UUID.randomUUID().toString();
String value = Thread.currentThread().getId() + : + uuid;
try { while (!redisTemplate.execute(lockRedisScript, Collections.singletonList(userId + :syn), value, 5)) {
// 等待 1000 毫秒重试获得锁
logger.info(Thread.currentThread().getName() + : 尝试循环获取锁
TimeUnit.MILLISECONDS.sleep(1000);
}
logger.info(Thread.currentThread().getName() + : 获得锁 ----
// 开启守护线程
daemonThread = new DaemonThread(userId + :syn , redisTemplate);
Thread thread = new Thread(daemonThread);
thread.start();
// 业务逻辑执行 10 秒...
TimeUnit.MILLISECONDS.sleep(10000);
} catch (InterruptedException e) { logger.error( 异常 , e);
} finally {
// 使用 Lua 脚本：先判断是否是自己设置的锁，再执行删除
// key 存在, 当前值 = 期望值时, 删除 key;key 存在, 当前值!= 期望值时, 返回 0;
Long result = redisTemplate.execute(unlockRedisScript, Collections.singletonList(userId + :syn), value);
logger.info(redis 解锁:{} , RELEASE_SUCCESS.equals(result));
if (RELEASE_SUCCESS.equals(result)) { if (daemonThread != null) {
// 关闭守护线程
daemonThread.stop();
logger.info(Thread.currentThread().getName() + : 释放锁 ---
}
}
}
}其中 while 循环中加锁和 finally 中的释放锁都是基于 Lua 脚本来实现了。
Redis 锁的其他因素
除了上述实例，在使用 Redis 分布式锁时，还可以考虑以下情况及方案。
Redis 锁的不可重入
当线程在持有锁的情况下再次请求加锁，如果一个锁支持一个线程多次加锁，那么这个锁就是可重入的。如果一个不可重入锁被再次加锁，由于该锁已经被持有，再次加锁会失败。Redis 可通过对锁进行重入计数，加锁时加 1，解锁时减 1，当计数归 0 时释放锁。
可重入锁虽然高效但会增加代码的复杂性，这里就不举例说明了。
等待锁释放
有的业务场景，发现被锁则直接返回。但有的场景下，客户端需要等待锁释放然后去抢锁。上述示例就属于后者。针对等待锁释放也有两种方案：
客户端轮训：当未获得锁时，等待一段时间再重新获取，直到成功。上述示例就是基于这种方式实现的。这种方式的缺点也很明显，比较耗费服务器资源，当并发量大时会影响服务器的效率。
使用 Redis 的订阅发布功能：当获取锁失败时，订阅锁释放消息，获取锁成功后释放时，发送释放消息。
集群中的主备切换和脑裂
在 Redis 包含主从同步的集群部署方式中，如果主节点挂掉，从节点提升为主节点。如果客户端 A 在主节点加锁成功，指令还未同步到从节点，此时主节点挂掉，从节点升为主节点，新的主节点中没有锁的数据。这种情况下，客户端 B 就可能加锁成功，从而出现并发的场景。
当集群发生脑裂时，Redis master 节点跟 slave 节点和 sentinel 集群处于不同的网络分区。sentinel 集群无法感知到 master 的存在，会将 slave 节点提升为 master 节点，此时就会存在两个不同的 master 节点。从而也会导致并发问题的出现。Redis Cluster 集群部署方式同理。
到此，关于“如何理解分布式系统下基于 Redis 的分布式锁”的学习就结束了，希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习，快去试试吧！若想继续学习更多相关知识，请继续关注丸趣 TV 网站，丸趣 TV 小编会继续努力为大家带来更多实用的文章！
