前言

本文继续记录B站谷粒商城项目视频 P151-157 的内容，做到知识点的梳理和总结的作用。

一、本地缓存与分布式缓存

1.1 使用缓存

为了系统性能的提升，我们一般都会将部分数据放入缓存中，加速访问。而 db 承担数据落盘工作。
哪些数据适合放入缓存？

1. 即时性、数据一致性要求不高的
2. 访问量大且更新频率不高的数据（读多，写少）
   举例：电商类应用，商品分类，商品列表等适合缓存并加一个失效时间(根据数据更新频率来定)，后台如果发布一个商品，买家需要 5 分钟才能看到新的商品一般还是可以接受的。

伪代码

data = redisTemplate.opsForValue().get(redisKey);//从缓存加载数据
If(data == null){//缓存中没有则从数据库加载数据data = db.getDataFromDB(id);//保存到 cache 中redisTemplate.opsForValue().set(redisKey,data);
}
return data;

1.2 本地缓存

在单体项目中，我们可以使用 Map 集合存储数据作为项目的本地缓存，因为 Map 数据是存储与内存的，相比于数据库查询要从磁盘加载到内存有着更高的效率。

1.3 本地模式在分布式下的问题

但是在分布式情况下这种情况就不再适用了，每个微服务可能部署在多台机器上，每个机器上有各自的缓存 Map 对象，会导致数据不一致的问题。

1.4 分布式缓存

所以应该将数据缓存在同一个缓存中间件中，才能保证数据一致性问题

二、整合redis测试

2.1 引入依赖

<!-- 缓存中间件redis依赖-->
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

2.2 配置信息

spring:redis:host: 192.168.57.129port: 6379

2.3 测试

@Autowired
StringRedisTemplate redisTemplate;@Test
public void testRedis() {//存储redisTemplate.opsForValue().set("HELLO_REDIS", "SpringBoot!");//获取String value = redisTemplate.opsForValue().get("HELLO_REDIS");System.out.println(value);
}

三、改造三级分类业务

3.1 代码改造

@Override
public Map<String, List<Catelog2Vo>> getCatalogJson() {//给缓存中放json字符串，拿出的json字符串，还用逆转为能用的对象类型:【序列化与反序列化】/*** 1、空结果缓存：解决缓存穿透* 2、设置过期时间（加随机值）：解决缓存雪崩* 3、加锁：解决缓存击穿*///1、加入缓存逻辑,缓存中存的数据是json字符串。//JSON跨语言，跨平台兼容。String catalogJSON = redisTemplate.opsForValue().get("catalogJSON");if (StringUtils.isEmpty(catalogJSON)) {//2、缓存中没有,查询数据库//保证数据库查询完成以后,将数据放在redis中,这是一个原子操作。log.info("缓存不命中....将要查询数据库...");Map<String, List<Catelog2Vo>> catalogJsonFromDb = getCatalogJsonFromDB();String result = JSON.toJSONString(catalogJsonFromDb);redisTemplate.opsForValue().set("catalogJSON",result);}log.info("缓存命中....直接返回....");//转为我们指定的对象。return JSON.parseObject(catalogJSON, new TypeReference<Map<String, List<Catelog2Vo>>>() {});
}

四、高并发下缓存失效问题

4.1 缓存穿透

解决方案1：null 结果放入缓存，并加入短暂的过期时间

伪代码

//从缓存加载数据
data = redisTemplate.opsForValue().get(redisKey);
If(data == null){//缓存中没有则从数据库加载数据data = db.getDataFromDB(id);if(data == null) {//空结果保存到 cache 中redisTemplate.opsForValue().set(redisKey,null,300,TimeUnit.SECONDS);}else {//保存到 cache 中redisTemplate.opsForValue().set(redisKey,data);}
}
return data;

解决方案2：使用布隆过滤器
这种技术在缓存之前再加一层屏障，里面存储目前数据库中存在的所有key。当业务系统有查询请求的时候，首先去BloomFilter中查询该key是否存在。若不存在，则说明数据库中也不存在该数据，因此缓存都不要查了，直接返回null。若存在，则继续执行后续的流程，先前往缓存中查询，缓存中没有的话再前往数据库中的查询。伪代码如下：

String get(String key) {String value = redis.get(key);     if (value  == null) {if(!bloomfilter.mightContain(key)){//不存在则返回return null; }else{//可能存在则查数据库value = db.get(key); redis.set(key, value); }    }return value；
}

布隆过滤器示意图

4.2 缓存雪崩

4.3 缓存击穿

五、分布式下加锁

5.1 分布式锁示意图

本地锁，只能锁住当前进程，所以我们需要分布式锁。

5.2 锁的时序问题

@Override
public Map<String, List<Catelog2Vo>> getCatalogJson() {//给缓存中放json字符串，拿出的json字符串，还用逆转为能用的对象类型:【序列化与反序列化】/*** 1、空结果缓存：解决缓存穿透* 2、设置过期时间（加随机值）：解决缓存雪崩* 3、加锁：解决缓存击穿*///1、加入缓存逻辑,缓存中存的数据是json字符串。//JSON跨语言，跨平台兼容。String catalogJSON = redisTemplate.opsForValue().get("catalogJSON");if (StringUtils.isEmpty(catalogJSON)) {//2、缓存中没有,查询数据库//保证数据库查询完成以后,将数据放在redis中,这是一个原子操作。log.info("缓存不命中....将要查询数据库...");Map<String, List<Catelog2Vo>> catalogJsonFromDb = getCatalogJsonFromDB();return catalogJsonFromDb;}log.info("缓存命中....直接返回....");//转为我们指定的对象。return JSON.parseObject(catalogJSON, new TypeReference<Map<String, List<Catelog2Vo>>>() {});
}

查询数据库后将结果放入缓存，保证这是一个原子性操作，防止多个线程查询数据库而导致日志输出多个查询了数据库…

//从数据库查询并封装分类数据
public Map<String, List<Catelog2Vo>> getCatalogJsonFromDB() {//只要同一把锁,就能锁住需要这个锁的所有线程//synchronized (this)：springBoot所有组件在容器中都是单实例的//TODO 本地锁：synchronized JUC(Lock) 在分布式情况下只能使用分布式锁才能锁住资源synchronized (this) {//得到锁以后,我们应该再去缓存中确定一次,如果没有才需要继续查询String catalogJSON = redisTemplate.opsForValue().get("catalogJSON");if (!StringUtils.isEmpty(catalogJSON)) {return JSON.parseObject(catalogJSON, new TypeReference<Map<String, List<Catelog2Vo>>>() {});}log.info("查询了数据库....");//1、将数据库的多次查询变为一次,查询所有分类信息List<CategoryEntity> selectList = baseMapper.selectList(null);//1、查出所有1级分类List<CategoryEntity> level1Categorys = getParent_cid(selectList, 0L);//2、封装数据Map<String, List<Catelog2Vo>> parent_cid = level1Categorys.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、找当前二级分类的三级分类封装成voList<CategoryEntity> level3Catelog = getParent_cid(selectList, l2.getCatId());if (level3Catelog != null) {List<Catelog2Vo.Catelog3Vo> collect = level3Catelog.stream().map(l3 -> {//2、封装成指定格式Catelog2Vo.Catelog3Vo catelog3Vo = new Catelog2Vo.Catelog3Vo(l2.getCatId().toString(), l3.getCatId().toString(), l3.getName());return catelog3Vo;}).collect(Collectors.toList());catelog2Vo.setCatalog3List(collect);}return catelog2Vo;}).collect(Collectors.toList());}return catelog2Vos;}));String result = JSON.toJSONString(parent_cid);redisTemplate.opsForValue().set("catalogJSON",result,1,TimeUnit.DAYS);return parent_cid;}
}

压力测试结果：日志只输出一个查询了数据库…，表面只有一个线程查询了数据库。