22-09-20 西安谷粒商城（04）Redisson做分布式锁、布隆过滤器、AOP赋能、自定义注解做缓存管理、秒杀测试

Redisson

1、Redisson做分布式锁

分布式锁主流的实现方案：

基于数据库实现分布式锁
基于缓存（Redis），性能最高
基于Zookeeper，可靠性最高

Redisson是一个在Redis的基础上实现的Java驻内存数据网格，它不仅提供了一系列的分布式的Java常用对象，还提供了许多分布式服务。

Redisson的宗旨是促进使用者对Redis的关注分离（Separation of Concern），从而让使用者能够将精力更集中地放在处理业务逻辑上

1、依赖、配置

<dependency><groupId>org.redisson</groupId><artifactId>redisson</artifactId><version>3.11.2</version>
</dependency>

创建配置类

@Configuration
public class RedissonConfig {//初始化redis客户端对象注入到容器中@Beanpublic RedissonClient redissonClient(){Config config = new Config();// 可以用"rediss://"来启用SSL连接config.useSingleServer().setAddress("redis://192.168.2.108:6379");return Redisson.create(config);}
}

当然了还有一些常用配置，参考如下：

====================================

2、使用Redisson分布式锁

锁的名字就是锁的粒度，粒度越细越快

/*** 秒杀案例测试*/
@GetMapping("index/testlock")
public void testLock() {//只要锁的名称相同就是同一把锁  getlock（）获取一个可重入的锁RLock lock = this.redissonClient.getLock("lock");//阻塞等待获取锁，默认等待3秒lock.lock();//获取redis中num的值int num = Integer.parseInt(redisTemplate.opsForValue().get("num").toString());num++;//设置到redis中redisTemplate.opsForValue().set("num", num);//解锁lock.unlock();
}
1.锁名称相同就认为是同一把锁

2.redisson自动续期默认是30S，每10S会续期到30S,也就是每10S会进行喂狗操作
只要锁没有指定释放时间,每隔lockWatchdogTimeout/3 就会给锁续期,续满看门狗时间30S 
3.redisson底层的所有操作都依赖于lua脚本

4.加锁操作 + 过期时间操作能保证原子性获取锁 + 判断锁 + 删除锁也能保证原子性

获取锁的其他方式

除了最常用的可重入锁，还有公平锁和读写锁

//获取分布式公平锁
RLock fairLock = this.redissonClient.getFairLock("xxx");
//加锁
fairLock.lock();
//释放锁
fairLock.unlock();//获取分布式读写锁
RReadWriteLock rwLock = this.redissonClient.getReadWriteLock("xxx");
//读锁
rwLock.readLock().lock();
rwLock.readLock().unlock();
//写锁
rwLock.writeLock().lock();
rwLock.writeLock().unlock();

读写锁：允许一个写锁和多个读锁同时竞争

Redisson还通过加锁的方法提供了leaseTime的参数来指定加锁的时间。超过这个时间后锁便自动解开了。[可以防止死锁]

//可重入锁，最经常使用的锁
lock.lock();// 加锁以后10秒钟自动解锁
// 无需调用unlock方法手动解锁
lock.lock(10, TimeUnit.SECONDS);// 尝试加锁，最多等待100秒，上锁以后10秒自动解锁
boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);

测试一下，服务多实例启动，并给num设置为0

使用ab测试
ab -n 5000 -c 100 http://www.gmall.com/index/testlock
redis客户端结果，真不错

redlock算法

redis在分布式系统中保证分布式锁分布式数据安全的一种算法

安全属性（Safety property）: 独享（相互排斥）。在任意一个时刻，只有一个客户端持有锁。 setnx

活性A(Liveness property A): 无死锁。即便持有锁的客户端崩溃（crashed)或者网络被分裂（gets partitioned)，锁仍然可以被获取。设置过期时间

活性B(Liveness property B): 容错。只要大部分Redis节点都活着，客户端就可以获取和释放锁.

红锁：多个锁组成的锁，超过一半成功才代表获取到锁

2、Redisson分布式锁解决缓存穿透

热点key单个或者批量同时失效，大量的请求缓存查询失败，会去查询数据库。线程处理请求的时间变长，服务器并发能力下降，可能导致服务器宕机

解决办法：只让一个请求线程查询数据库，然后设置到缓存中其他的请求以后走缓存

    @Overridepublic List<CategoryEntity> queryLvl2CategoriesWithSub(Long pid) {//1.先查询缓存String key = "idx:cache:cates:" + pid;Object obj = redisTemplate.opsForValue().get(key);//缓存有的话直接返回if (obj != null) {return (List<CategoryEntity>) obj;}//2.缓存没有,加锁控制只让一个线程查询数据库的数据RLock lock = redissonClient.getLock("cates:lock");lock.lock(30l, TimeUnit.SECONDS);try {// 双查：解决并发多个等待获取锁查询数据库数据的线程每个都查询数据库：// 再次查询缓存 如果有缓存直接返回obj = redisTemplate.opsForValue().get(key);if (obj != null) {return (List<CategoryEntity>) obj;}//-只有第一个线程才会这么通过，远程服务调用查询2/3级分类ResponseVo<List<CategoryEntity>> listResponseVo = pmsFeign.queryCategoriesWithSub(pid);//3.将查询到的值设置到redis缓存中long ttl = 1800 + new Random().nextInt(200);//校验空值-解决缓存穿透问题if (CollectionUtils.isEmpty(listResponseVo.getData())) {//如果是空值，ttl有效期就设置的短一些ttl = new Random().nextInt(500);}redisTemplate.opsForValue().set(key, listResponseVo.getData(), ttl, TimeUnit.SECONDS);return listResponseVo.getData();} catch (Exception e) {e.printStackTrace();} finally {lock.unlock();}return null;}

在首页上触发鼠标移动事件，去查询2/3级分类

redis数据库中：

3、自定义注解@GmallCache

@Target(ElementType.METHOD)
@Documented
@Retention(RetentionPolicy.RUNTIME)
public @interface GmallCache {// 自定义注解需要管理的参数：缓存的key、锁的名称(击穿)、缓存过期时间、随机过期时间(雪崩)String key() default "cache";String lockName() default "lock";int timeout() default 5*60;//单位秒int random() default 2*60;//单位秒
}

此时我们就可以这么使用了，对缓存管理抽取，把上面那么长的方法简化到极致

@Override
@GmallCache(key = "idx:cache:cates" ,lockName = "cates:lock",timeout = 30*60 , random = 10*60)
public List<CategoryEntity> queryLvl2CategoriesWithSub(Long pid) {//远程服务调用查询2/3级分类ResponseVo<List<CategoryEntity>> listResponseVo = pmsFeign.queryCategoriesWithSub(pid);return listResponseVo.getData();
}

但是我们还得使用aop赋能，通过切面给注解添加功能

4、AOP使用【redis缓存管理+分布式锁】

获取标注@GmallCache的方法信息以及该方法上注解信息

编写一个切面类，咱不使用切入点表达式了，使用@annotation 对指定注解进行切面通知

@Aspect
@Component
public class GmallCacheAspect {//环绕通知: 对使用自定义注解GmallCache的方法进行增强//@annotation 对指定注解进行切面通知@Around("@annotation(com.atguigu.gmall.index.aspect.GmallCache)")public Object around(ProceedingJoinPoint joinPoint) throws Throwable {//切入点： 可以获取到切入点方法对象 和参数等信息MethodSignature signature = (MethodSignature) joinPoint.getSignature();//获取切入点方法信息Method method = signature.getMethod();//获取目标方法的对象Class returnType = signature.getReturnType();//1.获取目标方法的返回值类型System.out.println("目标方法的返回值类型: "+returnType.getName());//2、获取目标方法的参数列表Object[] args = joinPoint.getArgs();System.out.println("目标方法的参数: "+args[0].toString());//3、获取目标方法上的注解对象：获取注解对象中的 缓存key  lockkey  过期时间...GmallCache gmallCache = method.getAnnotation(GmallCache.class);String key = gmallCache.key();String lockName = gmallCache.lockName();int timeout = gmallCache.timeout();int random = gmallCache.random();System.out.println("目标方法上的注解: "+"key="+key+",lockName="+lockName+",timeout="+timeout+",random="+random);//4、执行目标对象方法:查询数据库中的二级分类和子集合数据Object result = joinPoint.proceed(args);return result;}
}

环绕通知:@Around，目标方法的执行需要我们手动调用，在它的前后进行扩展

2021/10/30 北京 spring【3】静态代理，动态代理、 AOP面向切面编程_￡小羽毛的博客-CSDN博客

浏览器访问： http://localhost:18087/index/cates/2

控制台打印：

===========

使用aop赋能完成缓存管理

在上面获取到那么多的信息之后，我们就可以对该方法做缓存管理了

@Aspect
@Component
public class GmallCacheAspect {@AutowiredRedisTemplate redisTemplate;@AutowiredRedissonClient redissonClient;//环绕通知: 对使用自定义注解GmallCache的方法进行增强//@annotation 对指定注解进行切面通知@Around("@annotation(com.atguigu.gmall.index.aspect.GmallCache)")public Object around(ProceedingJoinPoint joinPoint) throws Throwable {//切入点： 可以获取到切入点方法对象 和参数等信息MethodSignature signature = (MethodSignature) joinPoint.getSignature();//获取切入点方法信息Method method = signature.getMethod();//获取目标方法的对象Class returnType = signature.getReturnType();//1.获取目标方法的返回值类型System.out.println("目标方法的返回值类型: "+returnType.getName());//2、获取目标方法的参数列表Object[] args = joinPoint.getArgs();System.out.println("目标方法的参数: "+args[0].toString());//3、获取目标方法上的注解对象：获取注解对象中的 缓存key  lockkey  过期时间...GmallCache gmallCache = method.getAnnotation(GmallCache.class);String key = gmallCache.key();String lockName = gmallCache.lockName();int timeout = gmallCache.timeout();int random = gmallCache.random();System.out.println("目标方法上的注解: "+"key="+key+",lockName="+lockName+",timeout="+timeout+",random="+random);//4，判断是否存在缓存，有缓存则直接返回String cacheKey = key;if(ArrayUtils.isNotEmpty(args)){cacheKey = key+":"+ StringUtils.join(args,"-");lockName = lockName+":"+StringUtils.join(args,"-");}//使用布隆过滤器去决定是否查询缓存。。 todo// redisTemplate配置过键和值的序列化器，所以返回的Object真实类型就是原数据自己的类型Object obj = redisTemplate.opsForValue().get(cacheKey);if(obj!=null){return obj;}//5、缓存不存在，执行目标方法:查询数据库中的二级分类和子集合数据//分布式锁：解决雪崩RLock lock = redissonClient.getLock(lockName);lock.lock();//再次判断是否有缓存obj = redisTemplate.opsForValue().get(cacheKey);if(obj!=null){lock.unlock();//查询到缓存后释放锁return obj;}try{Object result = joinPoint.proceed(args);long cacheTime = timeout+new Random().nextInt(random);if(result==null  || (result instanceof List && CollectionUtils.isEmpty((List)result))){//空值也存入到缓存中  时间稍短cacheTime = random;}//存入缓存redisTemplate.opsForValue().set(cacheKey , result , cacheTime, TimeUnit.SECONDS);return result;}finally {lock.unlock();}}
}

5、信号量和闭锁

在配置类中可以初始化信号量和闭锁

//初始化redis客户端对象注入到容器中
@Bean
public RedissonClient redissonClient(){Config config = new Config();// 可以用"rediss://"来启用SSL连接config.useSingleServer().setAddress("redis://192.168.2.108:6379");RedissonClient redissonClient = Redisson.create(config);//1初始化信号量RSemaphore sempahore = redissonClient.getSemaphore("sempahore");//设置资源数量sempahore.trySetPermits(1);//2 初始化闭锁RCountDownLatch cdl = redissonClient.getCountDownLatch("cdl");cdl.trySetCount(1);return redissonClient;
}

这里就测试一下闭锁。。

    @Autowiredprivate RedissonClient redissonClient;@ResponseBody@GetMapping("/index/testlock2")public ResponseVo testLock2(){//testLock2 的请求阻塞等待cdl 闭锁的值为0才继续执行RCountDownLatch cdl = redissonClient.getCountDownLatch("cdl");try {cdl.await();} catch (InterruptedException e) {e.printStackTrace();}System.out.println("testLock2终于执行了....");return ResponseVo.ok();}@ResponseBody@GetMapping("/index/testlock3")public ResponseVo testLock3(){RCountDownLatch cdl = redissonClient.getCountDownLatch("cdl");try {cdl.countDown();//倒计数-1} catch (Exception e) {e.printStackTrace();}System.out.println("testLock3执行了....");return ResponseVo.ok();}

访问testLock2方法：

浏览器效果

访问testlock3方法：

浏览器效果俩个效果都出来了。。。

6、秒杀测试

分布式锁可以解决秒杀超卖问题，但是一次只能有一个线程获取锁，执行秒杀的业务。高并发的场景下吞吐量低。改为使用分布式信号量

秒杀的思路图

秒杀的伪代码

管理员初始化，1001是商品skuId

布隆过滤器BloomFilter

1、布隆过滤器特征

布隆过滤器是一种数据结构，比较巧妙的概率型数据结构，特点是高效地插入和查询，可以用来告诉 “某样东西一定不存在或者可能存在”。

==================

布隆过滤器是一种牺牲准确率换取空间及时间效率的概率型数据结构，它的3个特征

布隆过滤器判定一个数据不存在，它就一定不存在

判定一个数据存在，它可能不存在（误判）

数据只能插入不能删除

布隆过滤器的优点：

• 时间复杂度低，增加和查询元素的时间复杂为O(N)，（N为哈希函数的个数，通常情况比较小）

• 保密性强，布隆过滤器不存储元素本身

• 存储空间小，如果允许存在一定的误判，布隆过滤器是非常节省空间的（相比其他数据结构如Set集合）

布隆过滤器的缺点：

• 有点一定的误判率，但是可以通过调整参数来降低

• 很难删除元素

2、布隆过滤器原理

参考文章：布隆过滤器原理以及解决redis穿透问题-KuangStudy-文章

其内部维护一个全为0的bit数组，需要说明的是，布隆过滤器有一个误判率的概念，误判率越低，则数组越长，所占空间越大。误判率越高则数组越小，所占的空间越小。

数据加入这个集合时：

假设，根据误判率，我们生成一个10位的bit数组，以及2个hash函数（(f_1,f_2)）

0代表不存在某个数据，1代表存在某个数据。

假设输入集合为((N_1,N_2)),经过计算(f_1(N_1))得到的数值得为2，(f_2(N_1))得到的数值为5，则将数组下标为2和下表为5的位置置为1，如下图所示

同理，经过计算(f_1(N_2))得到的数值得为3，(f_2(N_2))得到的数值为6，则将数组下标为3和下表为6的位置置为1，如下图所示

==========

数据查询过程

这个时候，我们有第三个数(N_3)，我们判断(N_3)在不在集合((N_1,N_2))中，就进行(f_1(N_3)，f_2(N_3))的计算
若值恰巧都位于上图的红色位置中，我们则认为，(N_3)在集合((N_1,N_2))中若值有一个不位于上图的红色位置中，我们则认为，(N_3)不在集合((N_1,N_2))中
以上就是布隆过滤器的计算原理

3、影响BloomFilter误判率因素

影响布隆过滤器误判率的因素，有两个：

布隆过滤器的bit数组长度

过小的布隆过滤器很快所有的 bit 位均为 1，那么查询任何值都会返回“可能存在”，起不到过滤的目的了。布隆过滤器的长度会直接影响误报率，布隆过滤器越长其误报率越小。

布隆过滤器的hash函数个数

个数越多则布隆过滤器 bit 位置位 1 的速度越快，且布隆过滤器的效率越低；但是如果太少的话，那我们的误报率会变高。

如何选择适合业务的哈希函数的个数（k）和bit数组的长度（m）值呢，公式如下：
k 为哈希函数个数，m 为布隆过滤器长度，n 为插入的元素个数，p 为误报率
布隆过滤器真实失误率p公式：

布隆过滤器不需要我们自己来实现，因为已经有很多成熟的实现方案：

Google的guava
redisson
redis插件官方地址：GitHub - RedisBloom/RedisBloom: Probabilistic Datatypes Module for Redis

4、布隆重建

删除困难，需要额外编写逻辑进行布隆重建

原理:将之前布隆对象删除,重新生成一个布隆对象即可，通过定时任务实现

5、谷歌guava的布隆过滤器

引入依赖

<dependency><groupId>com.google.guava</groupId><artifactId>guava</artifactId><version>30.1.1-jre</version>
</dependency>

测试类

@SpringBootTest //自动提供IOC容器
public class MybatisTest {BloomFilter<String> bloomFilter;@PostConstruct //布隆过滤器的初始化public void init(){//参数1： 指定将来存入布隆过滤器的 数据 类型+编码 (计算hash的算法)//参数2： 预期的元素个数//参数3： 误判率bloomFilter =BloomFilter.create(Funnels.stringFunnel(Charsets.UTF_8), 10, 0.3);//存入数据：使用多个hash算法为存入的数据计算 并对数组长度求余 设置到布隆过滤器的数组中bloomFilter.put("1");bloomFilter.put("2");bloomFilter.put("3");bloomFilter.put("4");bloomFilter.put("5");}@Testpublic void contextLoads() {System.out.println("1："+bloomFilter.mightContain("1"));System.out.println("2："+bloomFilter.mightContain("2"));System.out.println("3："+bloomFilter.mightContain("3"));System.out.println("4："+bloomFilter.mightContain("4"));System.out.println("5："+bloomFilter.mightContain("5"));System.out.println("6："+bloomFilter.mightContain("6"));System.out.println("7："+bloomFilter.mightContain("7"));System.out.println("8："+bloomFilter.mightContain("8"));System.out.println("9："+bloomFilter.mightContain("9"));System.out.println("10："+bloomFilter.mightContain("10"));System.out.println("11："+bloomFilter.mightContain("11"));System.out.println("12："+bloomFilter.mightContain("12"));System.out.println("13："+bloomFilter.mightContain("13"));System.out.println("14："+bloomFilter.mightContain("14"));System.out.println("15："+bloomFilter.mightContain("15"));System.out.println("16："+bloomFilter.mightContain("16"));}}

不存在的一定不存在认为存在的可能不存在，看看结果

6、redisson的布隆过滤器

依赖的话就是之前引入的这个

<dependency><groupId>org.redisson</groupId><artifactId>redisson</artifactId><version>3.11.2</version>
</dependency>

配置类配置

@Configuration
public class RedissonConfig {//初始化redis客户端对象注入到容器中@Beanpublic RedissonClient redissonClient(){Config config = new Config();// 可以用"rediss://"来启用SSL连接config.useSingleServer().setAddress("redis://192.168.2.108:6379");return Redisson.create(config);}
}

测试类中测试

@Autowired
RedissonClient redissonClient;
@Test
public void testRedissonBloom(){RBloomFilter<String> bloom = this.redissonClient.getBloomFilter("bloom");bloom.tryInit(10l, 0.3);bloom.add("1");bloom.add("2");bloom.add("3");bloom.add("4");bloom.add("5");System.out.println("1："+bloom.contains("1"));System.out.println("2："+bloom.contains("2"));System.out.println("3："+bloom.contains("3"));System.out.println("4："+bloom.contains("4"));System.out.println("5："+bloom.contains("5"));System.out.println("6："+bloom.contains("6"));System.out.println("7："+bloom.contains("7"));System.out.println("8："+bloom.contains("8"));System.out.println("9："+bloom.contains("9"));System.out.println("10："+bloom.contains("10"));System.out.println("11："+bloom.contains("11"));System.out.println("12："+bloom.contains("12"));System.out.println("13："+bloom.contains("13"));System.out.println("14："+bloom.contains("14"));System.out.println("15："+bloom.contains("15"));System.out.println("16："+bloom.contains("16"));
}

结果打印：看的出来确实没人家谷歌算法做得好

7、BloomFilter解决缓存穿透

使用布隆过滤器主要是为了解决Redis缓存穿透问题

布隆过滤器可以前置到查询缓存之前，但是没有必要，因为大部分请求缓存直接命中返回

上面使用Ridisson缓存管理的缓存穿透解决方案是：缓存空值，但是会导致 redis中内存占用过高

添加布隆过滤器的配置类：

这样在项目启动时就可以将所有的二级分类的id存入到布隆过滤器中

项目启动时可以将存在的数据查询出来使用布隆过滤器的多种hash算法，每个算法计算数据的hash值
得到了多个值再和非常长的数组的长度进行求余将多个hash算法计算后hash值%数组的余存到数组中

@Configuration
public class BloomFilterConfig {@Autowiredprivate RedissonClient redissonClient;@Autowiredprivate GmallPmsFeign gmallPmsFeign;@Beanpublic RBloomFilter rBloomFilter(){// 初始化布隆过滤器RBloomFilter<String> bloomfilter = this.redissonClient.getBloomFilter("bloom:cates");//参数1：预期的元素个数  参数2：误判率bloomfilter.tryInit(50l, 0.03);//远程服务调用查询所有的二级分类idResponseVo<List<CategoryEntity>> listResponseVo = this.gmallPmsFeign.queryCategory(0l);List<CategoryEntity> categoryEntities = listResponseVo.getData();if (!CollectionUtils.isEmpty(categoryEntities)){//每个二级分类的分类id存入到布隆过滤器categoryEntities.forEach(categoryEntity -> {bloomfilter.add(categoryEntity.getId().toString());});}return bloomfilter;}
}

修改“AOP缓存管理”封装的代码：

如果传入二级分类的id了，布隆过滤器说不存在那就是一定不存在，所以直接return null，不用再去查询缓存了

if(ArrayUtils.isNotEmpty(args)){//使用布隆过滤器判断查询的数据是否存在：//cid查询它的二级分类： cid必须存在 并且它有二级分类集合  这个cid就有数据// 项目启动时 可以将存在并且有二级分类的cid存入到布隆过滤器中RBloomFilter<Object> bloomFilter = redissonClient.getBloomFilter("bloom:cates");boolean contains = bloomFilter.contains(args[0].toString());if(!contains){//缓存一定不存在log.info("bloomfilter判断数据不存在：{}" ,args[0].toString());return null;}cacheKey = key+":"+StringUtils.join(args,"-");lockName = lockName+":"+StringUtils.join(args,"-");
}

总结：缓存穿透的解决方案

1、布隆过滤器过滤大部分的空值请求，但是它有误判率
2、redis缓存空值