谷粒商城|缓存使用-＞本地锁-＞分布式锁数据库|分布式锁|本地锁|优化

1.为了系统性能的提升，我们一般都会将部分数据放入缓存中，加速访问，而数据库承担数据罗盘工作。
那么我们怎么将哪些数据放入缓存呢？或者说哪些数据比较适合放入缓存呢？
①即时性，数据一致性要求不高的（通俗点说就是数据就算不马上更新，对用户几乎也是没有影响的。）
②访问量大且更新频率不高的数据（读多写少的数据），例如电商的商品，访问量很大，给商品加一个缓存并加一个失效时间（根据数据更新频率来定），举个栗子:后台发布一个商品，用户在几分钟之后看到还是可以接受的。

缓存使用流程

文章图片

请求一个数据先去看缓存中有没有，如果缓存命中的话，则直接返回结果，如果没有命中的话，就去db里查，查完之后将数据放入缓存。
注意：在开发中，凡是放入缓存中的数据应该指定过期时间，使其可以在系统没有主动更新数据也能自动触发数据加载进缓存的流程。

伪代码

data = https://www.it610.com/article/cache.load(id); //从缓存加载数据 if(data == null){ data = db.load(id)//从数据库中查 cache.put(id,data); //查完之后保存到缓存中 }

本地缓存：

文章图片

比如图中的三个方法所获得的数据被频繁获取我们就将他放入缓存中，下次在由请求的时候，就直接从缓冲中拿，避免了从数据库中查。

本地缓存模式在分布式下的问题。

文章图片

在分布式系统中，商品服务可能部署好多个服务器，每一个商品服务都带一个本地缓存，那么问题就暴露出来了。
问题1：
如果getCategorys方法通过网关被负载均衡到第一个服务器了，第一个服务器没有这个数据，那么从数据库查出来后，就会被放入缓存。如果下一次getCategorys被负载均衡到其他的服务器，由于本地缓存是存在于那个商品服务所在的服务器上的，那么其他的服务器发现自己的缓存中并没有，还是会进行查数据库，第三次，第四次…还是很大概率出现这样的问题。
问题2：
上边的问题只是会出现效率上的问题，没有出现获取数据错误的问题。
但是如果收到一个修改商品价格的请求，这个请求被负载均衡到第一个服务器，然后成功的进行了修改，并且陈宫的更新了缓存。但是，第二个，第三个，第四个…服务器的缓存并没有进行更新，如果在有获取商品价格的请求来到的之后，他被负载均衡到了1号服务器之外的服务器（很大概率！），由于之前缓存中有这个商品的信息，直接就从缓冲中拿到了该商品的信息，但是！这个服务器的缓存并没有更新！也就是说，用户拿到了一个错误的数据，出现了数据一致性的问题。

所在本地缓存模式在分布式系统中不能用！

本地缓存模式出现的问题根源在哪？
是不是一个服务器一个缓存导致的？
所以我们让所有的服务器共用一个缓存。

文章图片

由于所有的服务共用一个缓存，也就是说每个服务进行更新缓存的时候，都是更新的大家共用的那一个缓存，也就不会出现其他服务缓存未更新以及后边再有请求到来的时候出现数据不一致的问题了。

而共用缓冲可以用缓冲中间件（redis…）来实现，还可以进行集群工作。

加锁解决缓存击穿问题。
缓存击穿就是一个热点数据刚好过期，恰巧这个时间点超高并发访问这个热点数据，由于已经过期了，那么这些超高并发就会落到db，我们称为缓存击穿。
解决方法是：超高并发下，只第一个进去的线程去查，查到后放入缓存，在返回数据，将其他的线程所在外面，查到以后释放锁，其他人获取到锁，先查缓存，由于已经被第一个线程放入到了缓存中，这样后来的线程就会直接命中缓存了。

加锁的位置：

分析一下：
高并发下，在一个时间点会有很多个线程通过判断缓存不命中的条件，进入到从db中查数据的方法。
所以我们要进行加锁，把从db中查数据的方法锁住，让那些通过条件的线程，让第一个进入db中查数据的方法的线程，查完数据后放入到缓存中，一定要在锁住的方法中放，这样这些通过缓存不命中条件的线程，在第一个线程释放掉锁后，他们会依次拿到锁，但是他们会命中缓存，从而不会去查数据库。至于后边的那些进程，他们都不会通过缓存不命中的条件，而是直接命中缓存。如下图

文章图片

先进来的10w线程，第一个拿到锁的线程会放入缓存中，然后释放掉锁，接下来的99999个线程都会依次执行拿锁，命中缓存，返回，释放锁这四个操作。后边的90w数据直接就会命中缓存。
在从数据库中查数据方法输出"去数据库中查…"这句话，
在从数据库中查数据方法中命中缓存时出书"getCatalogJsonFromDB方法命中缓存…直接返回…"这句话，
直接命中缓存的输出"命中缓存…直接返回…"这句话
压测一下

文章图片

去数据库中查只打印了一次说明我们加锁位置正确，

文章图片

打印这条语句的这些线程则是和我们之前说的那10w个线程一样的线程。

文章图片

而这些线程就是和我们之前说的90w线程一样的线程。

但是本地锁只能当前服务，如果是多个服务的话，本地锁就不好用了。每个服务都会查一遍数据库。
压测一下

文章图片

的确是锁不住的，每个服务都会查一遍数据库，虽然服务数绝对了查数据库数，但是如果这个查询非常耗时的情况下，对数据库和用户的响应很不好。
所以我们要用分布式锁。

分布式锁原理：

由于本地锁拿到的锁是每个服务都有的锁（因为this那个类一个服务一个），也就说这个锁只能锁住当前服务的其他线程，要想所有线程拿到的是同一把锁，就要用到分布式锁。

那么所有的服务共同拥有的或者说共同使用的有什么东西呢？

缓存！对就是缓存！
redis中可以通过set key value NX EX 过期时间

文章图片

所以我们实现如下

文章图片

压测一下

文章图片

终于成功了！！！！
4个服务只在10001的端口里查了一次db！！

controller调的getCatalogJson方法

@Override public Map> getCatalogJson() {/** * 要想完美使用缓存 * 1.null值进行缓存：解决缓存穿透 * 2.对缓存中的数据设置一个随机的过期时间：解决缓存雪崩 * 3.对热点数据进行加锁：解决缓存击穿 */ //先去缓存中查 String catalogJSON = cache.opsForValue().get("catalogJSON"); //没有的话从缓存中查并且放入到缓存中 if (StringUtils.isEmpty(catalogJSON)) {Map> catalogJsonFromDB = getCatalogJsonFromDbWithRedisLock(); return catalogJsonFromDB; } else {System.out.println("命中缓存....直接返回..."); //有的话查出来反序列化然后返回 Map> result = JSON. parseObject(catalogJSON, new TypeReference>>() {}); return result; } }

getCatalogJsonFromDbWithRedisLock的方法

public Map> getCatalogJsonFromDbWithRedisLock() {//占好位置的同时一定要设置过期时间，如果占好位置后,执行完了业务逻辑，要删除锁的时候要断电了，就造成死锁了 //所以为了防止断电宕机我们要设置锁的过期时间 String uuid = UUID.randomUUID().toString(); Boolean lock = cache.opsForValue().setIfAbsent("lock", uuid, 3000, TimeUnit.SECONDS); if (lock) { //拿到锁 //执行业务 System.out.println("获取分布式锁成功......"); Map> catalogJsonFromDb; try {catalogJsonFromDb = getCatalogJsonFromDb(); } finally { //执行完了进行原子删锁 //lua脚本 String script = "if redis.call('get',KEYS[1]) == ARGV[1] then return redis.call('del',KEYS[1])elsereturn 0 end"; //原子删锁 cache.execute(new DefaultRedisScript(script, Long.class), Arrays.asList("lock"), uuid); } ////获取值进行对比成功则删除——>原子操作 //if (uuid.equals(cache.opsForValue().get("lock"))){//会导致释放别人锁的问题 //cache.delete("lock"); //}return catalogJsonFromDb; } else {//没拿到锁则一直尝试拿（自旋） System.out.println("获取分布式锁失败....正在等待重试.........."); try {//防止把栈挤炸睡一下在请求 Thread.sleep(300); } catch (InterruptedException e) {e.printStackTrace(); } return getCatalogJsonFromDbWithRedisLock(); }}

从db获取数据的方法

public Map> getCatalogJsonFromDb() {String catalogJSON = cache.opsForValue().get("catalogJSON"); //如果不空的话直接返回 if (!StringUtils.isEmpty(catalogJSON)) {//有的话查出来反序列化然后返回 Map> result = JSON. parseObject(catalogJSON, new TypeReference>>() {}); return result; }System.out.println("去数据库中查........"); //获取所有的分类 List all = this.list(); //获取所有一级分类List oneLevelCategory = getCatalogEntityByParentId(all, 0L); Map> res = oneLevelCategory.stream().collect(Collectors.toMap( k -> {return k.getCatId().toString(); }, v -> {//QueryWrapper list2Wrapper = new QueryWrapper().eq("parent_cid", v.getCatId()); //List level2Catalogys = this.list(list2Wrapper); //优化：在所有的分类中parent_id是它的 //查出所有二级分类 List level2Catalogys = getCatalogEntityByParentId(all, v.getCatId()); List catelog2VoList = null; //并进行封装 if (level2Catalogys != null) {catelog2VoList = level2Catalogys.stream().map(l2 -> {Catelog2Vo catelog2Vo = new Catelog2Vo(); catelog2Vo.setCatalog1Id(l2.getParentCid().toString()); catelog2Vo.setId(l2.getCatId().toString()); catelog2Vo.setName(l2.getName()); //查出三级分类 //List level3Catelogys = this.list(new QueryWrapper().eq("parent_cid", l2.getCatId())); List level3Catelogys = getCatalogEntityByParentId(all, l2.getCatId()); //封装好了 if (level3Catelogys != null) {List collect = level3Catelogys.stream().map(l3 -> {Catelog2Vo.Catelog3Vo catelog3Vo = new Catelog2Vo.Catelog3Vo(); catelog3Vo.setCatalog2Id(l3.getParentCid().toString()); catelog3Vo.setId(l3.getCatId().toString()); catelog3Vo.setName(l3.getName()); return catelog3Vo; }).collect(Collectors.toList()); catelog2Vo.setCatalog3List(collect); } return catelog2Vo; }).collect(Collectors.toList()); }return catelog2VoList; })); String jsonString = JSON.toJSONString(res); //第一个拿到锁的线程返回之前放入到缓存（一定要在锁住！） cache.opsForValue().set("catalogJSON", jsonString); return res; }