day07-优惠券秒杀03
功能03-优惠券秒杀03
4.功能03-优惠券秒杀
4.6Redisson的分布式锁
4.6.1基于setnx实现的分布式锁问题
我们在4.5自己实现的分布式锁，主要使用的是redis的setnx命令，它仍存在如下问题：
4.6.2Redisson基本介绍
Redisson是一个在Redis基础上实现的Java驻内存数据网格（In-Memory Data Grid）。它不仅提供了一系列的分布式的Java常用对象，还提供了许多分布式服务，其中就包括了各种分布式锁的实现。
一句话：Redisson是一个在Redis基础上实现的分布式工具的集合。
据Redisson官网的介绍，Redisson是一个Java Redis客户端，与Spring 提供给我们的 RedisTemplate 工具没有本质的区别，可以把它看做是一个功能更强大的客户端
官网地址： https://redisson.org
GitHub地址： https://github.com/redisson/redisson
中文文档：目录 · redisson/redisson Wiki (github.com)
4.6.3Redisson快速入门
代码实现
（1）修改pom.xml，添加依赖
<!--redisson-->
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.13.6</version>
</dependency>
（2）配置Redisson
package com.hmdp.config;
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;
/**
* @author 李
* @version 1.0
*/
@Configuration
public class RedissonConfig {
@Bean
public RedissonClient redissonClient() {
//配置
Config config = new Config();
//redis单节点模式，设置redis服务器的地址，端口，密码
config.useSingleServer().setAddress("redis://127.0.0.1:6379").setPassword("123456");
//创建RedissonClient对象
return Redisson.create(config);
}
}
配置了之后，就可以在任意地方去使用Redisson了：比如说去改造之前的业务，使用Redisson的分布式锁
（3）修改VoucherOrderServiceImpl.java，使用Redisson的分布式锁
注入RedissonClient对象：
使用RedissonClient提供的锁：
（4）使用jemeter测试
分别向端口为8081、8082的服务器发送200个请求（使用同一个用户的token）
数据库中只下了一单：
说明解决了集群下的一人一单问题。
4.6.4Redisson可重入锁原理（Reentrant Lock）
可重入锁：字面意思是“可以重新进入的锁”，即允许同一个线程多次获取同一把锁。
比如一个递归函数里有加锁操作，递归过程中这个锁会阻塞自己吗？如果不会，那么这个锁就是可重入锁（因为这个原因可重入锁也叫做递归锁）。
Lock锁借助于底层一个voaltile的state变量来记录重入状态。如果当前没有线程持有这把锁，那么state=0，假如有线程持有这把锁，那么state=1，如果持有这把锁的线程再次持有这把锁，那么state就会+1 。
对于synchronized而言，它在c语言代码中会有一个count，原理和state类似，也是重入一次就加一，释放一次就减一 ，直到减少成零时，表示当前这把锁没有被人持有。
Redisson也支持可重入锁。Redisson在分布式锁中，采用redis的hash结构用来存储锁，其中key表示这把锁是否存在，用field表示当前这把锁被哪个线程持有，value记录重入的次数（锁计数）。当获取锁的线程释放锁前，先对锁计数-1，然后判断锁计数0，如果是0，就释放锁。
使用Redis的string类型的setnx命令，可以实现互斥性，ex可以设置过期时间。但如果使用hash结构，该结构中没有类似的组合命令，因此只能将之前的逻辑拆开。先判断是否存在，然后手动设置过期时间，逻辑如下：
可以看到，无论是获取锁还是释放锁，都比使用setnx实现的分布式锁复杂得多，而且实现需要有多个步骤。
因此，需要采用lua脚本来确保获取锁和释放锁的原子性：
- 获取锁的lua脚本
local key = KEYS[1]; -- 锁的key
local threadId = ARGV[1]; -- 线程唯一标识
local releaseTime = ARGV[2]; -- 锁的自动释放时间
-- 判断是否存在
-- 锁不存在
if(redis.call('exists', key) == 0) then
-- 不存在, 获取锁
redis.call('hset', key, threadId, '1');
-- 设置有效期
redis.call('expire', key, releaseTime);
return 1; -- 返回结果
end;
-- 锁已经存在，判断threadId是否是自己
if(redis.call('hexists', key, threadId) == 1) then
-- 如果是自己, 获取锁，重入次数+1
redis.call('hincrby', key, threadId, '1'); -- hincrby命令是对哈希表指定的field对应的value增长指定步长
-- 重新设置有效期
redis.call('expire', key, releaseTime);
return 1; -- 返回结果
end;
return 0; -- 代码走到这里,说明获取锁的不是自己，获取锁失败
- 释放锁的脚本
local key = KEYS[1]; -- 锁的key
local threadId = ARGV[1]; -- 线程唯一标识
local releaseTime = ARGV[2]; -- 锁的自动释放时间
-- 判断当前锁是否还是被自己持有
if (redis.call('HEXISTS', key, threadId) == 0) then
return nil; -- 如果已经不是自己，则直接返回，不进行操作
end;
-- 是自己的锁，则重入次数-1
local count = redis.call('HINCRBY', key, threadId, -1);
-- 然后判断重入次数是否已经为0
if (count > 0) then
-- 大于0，说明不能释放锁，重置有效期然后返回
redis.call('EXPIRE', key, releaseTime);
return nil;
else -- 等于0，说明可以释放锁，直接删除
redis.call('DEL', key);
return nil;
end;
代码测试
我们来测试一下Redisson的可重入锁：
package com.hmdp;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.boot.test.context.SpringBootTest;
import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;
/**
* @author 李
* @version 1.0
*/
@Slf4j
@SpringBootTest
class RedissonTest {
@Resource
private RedissonClient redissonClient;
private RLock lock;
@BeforeEach
void setUp() {
lock = redissonClient.getLock("order");
}
@Test
void method1() throws InterruptedException {
// 尝试获取锁
boolean isLock = lock.tryLock(1L, TimeUnit.SECONDS);
if (!isLock) {
log.error("获取锁失败 .... 1");
return;
}
try {
log.info("获取锁成功 .... 1");
method2();
log.info("开始执行业务 ... 1");
} finally {
log.warn("准备释放锁 .... 1");
lock.unlock();
}
}
void method2() {
// 尝试获取锁
boolean isLock = lock.tryLock();
if (!isLock) {
log.error("获取锁失败 .... 2");
return;
}
try {
log.info("获取锁成功 .... 2");
log.info("开始执行业务 ... 2");
} finally {
log.warn("准备释放锁 .... 2");
lock.unlock();
}
}
}
在method1()的boolean isLock = lock.tryLock(1L, TimeUnit.SECONDS);旁打上断点：
点击step over，显示获取锁成功：
打开redis，可以看到对应的hash数据，value记录的是线程重入锁的次数，此时value=1：
当前线程在method1()中调用method2()后，在method2()中重新获取锁，此时value记录的次数+1，value=2：
当method2()释放锁的时候，锁重入次数-1，value=1：
当执行到method1()释放锁的时候，锁重入次数-1，此时发现锁重入次数value=0，因此删除对应的key，真正释放锁。
我们进入RedissonLock的源码，发现里面也写了相关的lua脚本，这里的脚本和上面我们写的基本一致：
获取锁的脚本：
释放锁的脚本：
4.6.5Redisson的锁重试和WatchDog机制
（1）为什么需要WatchDog机制？
如果拿到分布式锁的节点宕机，且这个锁正好处于锁住的状态时，就会出现死锁问题。为了避免这种情况的发生，我们通常都会给锁设置一个过期时间。但随之而来又产生了新的问题：假如一个线程拿到了锁并设置了30s超时，但在30s后这个线程的业务没有执行完毕，锁已经超时释放了。可能会导致其他线程抢到锁，然后出现多线程并发的问题。
为了解决这种两难的境地：Redisson提供了watch dog 自动延期机制。
（2）WatchDog的自动延期机制
Redisson提供了一个监控锁的看门狗，它的作用是在Redisson实例被关闭前，不断的延长锁的有效期。也就是说，如果一个拿到锁的线程一直没有完成逻辑，那么看门狗会帮助线程不断的延长锁超时时间，锁不会因为超时而被释放。如果获取到分布式锁的节点宕机了，看门狗就无法延长锁的有效期，也避免了死锁的可能。
watchDog 只有在未指定加锁时间（leaseTime）时才会生效
默认情况下，看门狗的续期时间是30s，也可以通过修改Config.lockWatchdogTimeout来另行指定。另外Redisson 还提供了可以指定leaseTime参数的加锁方法来指定加锁的时间。超过这个时间后锁便自动解开了，不会延长锁的有效期。
（3）锁重试机制：利用信号量和PubSub功能实现等待、唤醒，获取锁失败的重试机制
（4）Redisson实例获取锁和释放锁的流程
（4.1）获取锁的逻辑：
- 首先去尝试获取锁，如果返回的ttl为null，则说明成功获取锁，然后需要判断是否走看门狗机制：
- 如果我们自己设置了leaseTime，就不会开启watchDog机制，直接返回true；
- 如果设置的leaseTime=-1，则开启watchDog，不停地更新有效期，然后返回true
- 如果返回的ttl不为null，说明获取锁失败。需要重试获取，在重试之前要先判断线程剩余的等待时间：
- 如果剩余等待时间<=0，说明该线程没有机会获取锁了，直接返回false；
- 如果如果剩余时间>0，就可以去尝试重新获取锁了。但是不是立即吃重试获取，需要去等待锁释放的信号：
- 如果在等待中，等待时间大于了剩余等待时间，则直接返回false；
- 如果收到了释放锁的信号，并且如果等待时间小于剩余等待时间，就重新开始尝试获取锁
重复上述所有步骤。最终线程要么成功获取锁，要么超时返回。
（4.2）释放锁的逻辑
尝试释放锁：
- 如果失败，记录异常，结束
- 如果成功，向等待的其他线程发送释放锁信息。然后取消watchDog机制，结束
4.6.6Redisson分布式锁总结
Redisson分布式锁原理：
- 可重入：利用hash结构记录线程id和重入次数
- 可重试：利用信号量和PubSub功能实现等待、唤醒，获取锁失败的重试机制
- 超时续约：利用watchDog，每隔一段时间（releaseTime/3），重置超时时间
4.6.7Redisson的联锁原理（multiLock）
上面我们已经介绍了Redisson分布式锁如何实现锁的可重入，锁获取时的重试，以及锁释放时间的自动续约。
现在来分析一下Redisson怎么解决主从一致性问题。要解决这个问题，主从一致性问题产生：
（1）主从一致性问题
为避免单节点的redis服务宕机，从而影响依赖于redis的业务的执行（如分布式锁），在实际开发中，我们往往会搭建Redis的主从模式。
什么叫做Redis的主从模式？
有多台Redis，将其中一台Redis服务器作为主节点，其他的作为从节点。一般主节点负责写入数据，从节点负责读取数据，当主节点服务器写入数据时会同步到从节点的服务器上。
但主从节点毕竟不是在同一台机器上，它们之间的数据同步会有一定的延时，主从一致性问题正是由于这样的延时而导致的：
假设有一个Java应用现在要来获取锁，它向主节点间发送了一个写命令：set lock thread1 nx ex 10，主节点上保存了这个锁的标识，然后主节点向从节点同步数据，但就在这时主节点宕机了。也就是说同步未完成，但主节点已经宕机了。
redis中的哨兵监控着整个集群的状态，它发现主节点宕机之后，首先断开与客户端的连接，然后在Redis Slave中选择一个当做新的主节点。
但是由于之前的主从同步未完成——也就是说锁已经丢失了。所以，此时我们的Java应用再来访问这个新的主节点时就会发现，锁已经没有了（锁失效了）。那么此时再有其他线程来获取锁也能获取成功，因此就会出现线程的并发安全问题——这就是主从一致性问题导致的锁失效问题。
（2）MultiLock锁
既然主从关系是一致性问题发生的原因，那么就不要使用主从节点了。我们将所有的节点都变为独立的redis节点，相互之间没有任何关系，都可以去做读写，每个节点的地位都是一样的。
此时我们获取锁的方式就改变了：获取锁时，要把加锁的逻辑写入到每一个独立的Redis节点上，只有所有的服务器都写入成功，此时才是加锁成功。
-
假设现在某个节点挂了，那么去获得锁的时候，只要有一个节点拿不到，都不能算是加锁成功，保证了加锁的可靠性。
-
因为没有主从节点，也就不会出现一致性问题；其次，随着redis节点的增多，redis可用性也提高了。
-
为了提高可用性，我们也可以对所有独立的Redis Node分别建立主从关系，让它们去做主从同步。
那么独立的Redis Node的主从关系会不会导致锁失效呢？
我们假设此时有一个Redis Node宕机了，并且它的数据没有同步到它的从节点。这时如果有其他线程想去获取锁，因为在其他Redis Node上不能拿到锁，因此不算是获取锁成功。也就是说，只要有任意一个节点在存活着，那么其他线程就不能趁机拿到锁，解决了锁失效问题。
这样的方案保留了既主从同步机制，确保了Redis集群高可用的特性，同时也避免了主从一致引发的锁失效问题。这套方案在Redisson中被称为MultiLock锁（联锁）：redisson中的MultiLock，可以把一组锁当作一个锁来加锁和释放。
那么MutiLock 加锁原理是什么呢？笔者画了一幅图来说明
当我们去设置了多个锁时，redission会将多个锁添加到一个集合中，然后用while循环去不停去尝试拿锁，但是会有一个总共的加锁时间，这个时间是用需要加锁的个数 * 1500ms ，假设有3个锁，那么时间就是4500ms，假设在这4500ms内，所有的锁都加锁成功， 那么此时才算是加锁成功，如果在4500ms有线程加锁失败，则会再次去进行重试