谷粒商城分布式高级（六）—— 缓存-缓存使用（整合redis & 缓存穿透/雪崩/击穿 & 本地锁）
作者：互联网
一、缓存
1、缓存使用
为了系统性能的提升，我们一般都会将部分数据放入缓存中，加速访问。而 db 承担数据落 盘工作。 哪些数据适合放入缓存？ （1）即时性、数据一致性要求不高的 （2）访问量大且更新频率不高的数据（读多，写少）
举例：电商类应用，商品分类，商品列表等适合缓存并加一个失效时间(根据数据更新频率来定)，后台如果发布一个商品，买家需要 5 分钟才能看到新的商品一般还是可以接受的。
本地缓存与分布式缓存
本地缓存：和微服务同一个进程。缺点：分布式时本地缓存不能共享
分布式缓存：缓存中间件，例如：redis
2、整合 redis 作为缓存
（1）安装 redis
参考：谷粒商城分布式基础（二）—— 环境搭建（基础篇）（虚拟机 & JDK & Maven & docker & mysql & redis & vue） 中的 “6、docker安装redis”
（2）gulimall-product 引入 redis-starter
<!--引入redis-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
（3）application.yml 配置 redis
（4）使用 RedisTemplate 操作 redis
@Autowired
StringRedisTemplate stringRedisTemplate;
@Test
public void testStringRedisTemplate(){
ValueOperations<String, String> ops = stringRedisTemplate.opsForValue();
//保存
ops.set("hello","word"+ UUID.randomUUID().toString());
//查询
String hello = ops.get("hello");
System.out.println("之前保存的数据是："+hello);
}
测试结果：
（5）切换使用 jedis 解决内存泄漏
目的：解决堆外内存溢出
修改 pom.xml,切换使用jedis
<!--引入redis-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
<exclusions>
<exclusion>
<groupId>io.lettuce</groupId>
<artifactId>lettuce-core</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
</dependency>
3、改造三级分类业务
修改 com.atguigu.gulimall.product.service.impl.CategoryServiceImpl 的 getCatalogJson 方法
@Autowired
StringRedisTemplate redisTemplate;
//TODO 产生堆外内存溢出：OutOfDirectMemoryError
//（1）springboot2.0以后默认使用lettuce作为操作redis客户端。它使用netty进行网络通信
//（2）lettuce的bug导致netty堆外内存溢出 -Xmx300m；netty如果没有指定堆外内存，默认使用-Xmx300m
// 可以通过-Dio.netty.maxDirectMemory进行设置
// 解决方案：不能使用-Dio.netty.maxDirectMemory只去调大堆外内存。
//（1）升级lettuce客户端
//（2）切换使用jedis
@Override
public Map<String, List<Catelog2Vo>> getCatalogJson() {
//给缓存中放json字符串，拿出的json字符串，还要逆转为能用的对象类型：【序列化与反序列化】
//1、加入缓存逻辑，缓存中存的数据是json字符串
//JSO跨语言、跨平台兼容
String catalogJson = redisTemplate.opsForValue().get("catalogJson");
if(StringUtils.isEmpty(catalogJson)){
//2、缓存中没有，查询数据库
Map<String, List<Catelog2Vo>> catalogJsonFromDb = getCatalogJsonFromDb();
//3、查到的数据再放入缓存，将对象转为json放到缓存中
String s = JSON.toJSONString(catalogJsonFromDb);
redisTemplate.opsForValue().set("catalogJson", s);
return catalogJsonFromDb;
}
//转为我们指定的对象
Map<String, List<Catelog2Vo>> result = JSON.parseObject(catalogJson, new TypeReference<Map<String, List<Catelog2Vo>>>(){});
return result;
}
/**
* 从数据库查询并封装分类数据
* @return
*/
public Map<String, List<Catelog2Vo>> getCatalogJsonFromDb() {
/**
* 1、将数据库的多次查询变为一次
*/
List<CategoryEntity> selectList = baseMapper.selectList(null);
//1、查询所有一级分类
List<CategoryEntity> level1Catagorys = getParent_cid(selectList, 0L);
//2、封装数据
Map<String, List<Catelog2Vo>> parent_cid = level1Catagorys.stream().collect(Collectors.toMap(k -> k.getCatId().toString(), v -> {
// 1、每一个的一级分类，查到这个以及分类的二级分类
List<CategoryEntity> categoryEntities = getParent_cid(selectList, v.getCatId());
//2、封装上面的结果
List<Catelog2Vo> catelog2Vos = null;
if (categoryEntities != null) {
catelog2Vos = categoryEntities.stream().map(l2 -> {
Catelog2Vo catelog2Vo = new Catelog2Vo(v.getCatId().toString(), null, l2.getCatId().toString(), l2.getName());
//1、找当前二级分类的三级分类封装成vo
List<CategoryEntity> level3Catalog = getParent_cid(selectList, l2.getCatId());
if(level3Catalog!=null){
List<Catelog2Vo.Category3Vo> collect = level3Catalog.stream().map(l3 -> {
//2、封装成指定格式
Catelog2Vo.Category3Vo category3Vo = new Catelog2Vo.Category3Vo(l2.getCatId().toString(), l3.getCatId().toString(), l3.getName());
return category3Vo;
}).collect(Collectors.toList());
catelog2Vo.setCatalog3List(collect);
}
return catelog2Vo;
}).collect(Collectors.toList());
}
return catelog2Vos;
}));
return parent_cid;
}
private List<CategoryEntity> getParent_cid( List<CategoryEntity> selectList, Long parent_cid) {
List<CategoryEntity> collect = selectList.stream().filter(item -> item.getParentCid() == parent_cid).collect(Collectors.toList());
return collect;
}
lettuce和jedis是操作redis的底层客户端，RedisTemplate是再次封装
二、缓存失效问题
先来解决大并发读情况下的缓存失效问题；
1、缓存穿透
（1）含义
指查询一个一定不存在的数据，由于缓存是不命中，将去查询数据库，但是数据库也无此记录，我们没有将这次查询的null写入缓存，
这将导致这个不 存在的数据每次请求都要到存储层去查询，失去了缓存的意义
（2）风险
利用不存在的数据进行攻击，数据库瞬时压力增大，最终导致崩溃
（2）解决方案
null结果缓存，并加入短暂过期时间
2、缓存雪崩
（1）含义
缓存雪崩是指在我们设置缓存时key采用了相同的过期时间， 导致缓存在某一时刻同时失效，请求全部转发到DB，DB瞬时 压力过重雪崩。
（2）解决方案：
原有的失效时间基础上增加一个随机值，比如1-5分钟随机，这 样每一个缓存的过期时间的重复率就会降低，就很难引发集体 失效的事件。
出现雪崩：降级 熔断
事前：尽量保证整个 redis 集群的高可用性，发现机器宕机尽快补上。选择合适的内存淘汰策略。
事中：本地ehcache缓存 + hystrix限流&降级，避免MySQL崩掉
事后：利用 redis 持久化机制保存的数据尽快恢复缓存
3、缓存击穿
（1）含义
对于一些设置了过期时间的key，如果这些key可能会在某些时间点被超高并发地访问，是一种非常“热点”的数据。
如果这个key在大量请求同时进来前正好失效，那么所有对 这个key的数据查询都落到db，我们称为缓存击穿。
（2）解决方案
加锁
大量并发只让一个去查，其他人等待，查到以后释放锁，其他 人获取到锁，先查缓存，就会有数据，不用去db
4、加锁解决缓存击穿问题
（1） 修改 com.atguigu.gulimall.product.service.impl.CategoryServiceImpl 的 getCatalogJson 方法
//TODO 产生堆外内存溢出：OutOfDirectMemoryError
//（1）springboot2.0以后默认使用lettuce作为操作redis客户端。它使用netty进行网络通信
//（2）lettuce的bug导致netty堆外内存溢出 -Xmx300m；netty如果没有指定堆外内存，默认使用-Xmx300m
// 可以通过-Dio.netty.maxDirectMemory进行设置
// 解决方案：不能使用-Dio.netty.maxDirectMemory只去调大堆外内存。
//（1）升级lettuce客户端
//（2）切换使用jedis
@Override
public Map<String, List<Catelog2Vo>> getCatalogJson() {
//给缓存中放json字符串，拿出的json字符串，还要逆转为能用的对象类型：【序列化与反序列化】
/**
* 1、空结果缓存：解决缓存穿透
* 2、设置过期时间（加随机值）：解决缓存雪崩
* 3、枷锁：解决缓存击穿
*/
//1、加入缓存逻辑，缓存中存的数据是json字符串
//JSO跨语言、跨平台兼容
String catalogJson = redisTemplate.opsForValue().get("catalogJson");
if(StringUtils.isEmpty(catalogJson)){
//2、缓存中没有，查询数据库
Map<String, List<Catelog2Vo>> catalogJsonFromDb = getCatalogJsonFromDb();
System.out.println("缓存不命中....将要查询数据库....");
return catalogJsonFromDb;
}
System.out.println("缓存命中....直接返回.....");
//转为我们指定的对象
Map<String, List<Catelog2Vo>> result = JSON.parseObject(catalogJson, new TypeReference<Map<String, List<Catelog2Vo>>>(){});
return result;
}
/**
* 从数据库查询并封装分类数据
* @return
*/
public Map<String, List<Catelog2Vo>> getCatalogJsonFromDb() {
//只要是同一把锁，就能锁住，需要这个锁的所有线程
//1、synchronized (this)：SprinBoot所有的组件在容器中都是单例的
//TODO 本地锁：synchronized，JUC(lock)，在分布式情况下，想要锁住所有，必须使用分布式锁
synchronized (this){
//得到锁以后，我们应该再去缓存中确定一次，如果没有才需要继续查询
String catalogJson = redisTemplate.opsForValue().get("catalogJson");
if(!StringUtils.isEmpty(catalogJson)){
//缓存不为null直接返回
Map<String, List<Catelog2Vo>> result = JSON.parseObject(catalogJson, new TypeReference<Map<String, List<Catelog2Vo>>>(){});
return result;
}
System.out.println("查询了数据库.....");
/**
* 1、将数据库的多次查询变为一次
*/
List<CategoryEntity> selectList = baseMapper.selectList(null);
//1、查询所有一级分类
List<CategoryEntity> level1Catagorys = getParent_cid(selectList, 0L);
//2、封装数据
Map<String, List<Catelog2Vo>> parent_cid = level1Catagorys.stream().collect(Collectors.toMap(k -> k.getCatId().toString(), v -> {
// 1、每一个的一级分类，查到这个以及分类的二级分类
List<CategoryEntity> categoryEntities = getParent_cid(selectList, v.getCatId());
//2、封装上面的结果
List<Catelog2Vo> catelog2Vos = null;
if (categoryEntities != null) {
catelog2Vos = categoryEntities.stream().map(l2 -> {
Catelog2Vo catelog2Vo = new Catelog2Vo(v.getCatId().toString(), null, l2.getCatId().toString(), l2.getName());
//1、找当前二级分类的三级分类封装成vo
List<CategoryEntity> level3Catalog = getParent_cid(selectList, l2.getCatId());
if(level3Catalog!=null){
List<Catelog2Vo.Category3Vo> collect = level3Catalog.stream().map(l3 -> {
//2、封装成指定格式
Catelog2Vo.Category3Vo category3Vo = new Catelog2Vo.Category3Vo(l2.getCatId().toString(), l3.getCatId().toString(), l3.getName());
return category3Vo;
}).collect(Collectors.toList());
catelog2Vo.setCatalog3List(collect);
}
return catelog2Vo;
}).collect(Collectors.toList());
}
return catelog2Vos;
}));
//3、查到的数据再放入缓存，将对象转为json放到缓存中
String s = JSON.toJSONString(parent_cid);
redisTemplate.opsForValue().set("catalogJson", s, 1, TimeUnit.DAYS);
return parent_cid;
}
}
private List<CategoryEntity> getParent_cid( List<CategoryEntity> selectList, Long parent_cid) {
List<CategoryEntity> collect = selectList.stream().filter(item -> item.getParentCid() == parent_cid).collect(Collectors.toList());
return collect;
}
锁时序问题：之前的逻辑是查缓存没有，然后取竞争锁查数据库，这样就造成多 次查数据库。
解决方法：竞争到锁后，再次确认缓存中没有，再去查数据库。
（2）jemeter 测试 100的并发 查看是否锁住，是否只查询了一次数据库
测试结果：
确认只查了一次，确实锁住了
5、本地锁在分布式下的问题
（1）模拟多个商品服务，分别copy三个商品服务，端口号依次设置为10001、10002、10003
分别启动这四个服务
（2）修改jemeter
（3）清除之前测试的缓存数据，启动测试，查看结果
结果发现，每一个服务的控制台都显示查询了一次数据库，也就是说只能锁住本地服务，不能解决分布式下的所问题
三、缓存数据一致性
1、保证一致性模式
（1）双写模式
（2）失效模式
（3）改进方法 1-分布式读写锁
分布式读写锁。读数据等待写数据整个操作完成 （4）改进方法 2-使用cananl
标签：缓存,return,cid,redis,List,collect,谷粒 来源： https://www.cnblogs.com/javahr/p/15725505.html