问题引入
缓存穿透
缓存穿透是指大并发下突然访问一个不存在的数据，导致一直去查询数据库，数据库瞬时压力增大，最终导致崩溃。
解决方法：设置具有过期时间的null数据
缓存雪崩
缓存雪崩是指我们会给缓存中的key value设置过期时间，假如100w条数据的过期时间是一样的，当数据过期的一瞬间突然100w的并发来，这时候数据库会崩溃。
解决方法：为不同的数据设置随机的过期时间，让他们不至于同时失效
缓存击穿
缓存击穿是指某一个热点数据在失效的一瞬间进来了高并发，所有对这个key的数据查询都落到了db上，我们称之为缓存击穿
解决方法：大量并发让一个人查，其他人等待，查到以后释放锁，其他人获取锁查看缓存有没有数据，没有再去db查
解决方案
本地锁
本地锁只能锁住this也就是当前实例对象，如果是单体应用不采用分布式的情况下是可以的。因为如果一个采用集群的方式部署，每个节点都有一把锁，并发进来的时候几把锁就可以就放进来了几个进程。
使用本地锁要注意在锁中就要进行数据的缓存，不然的话当锁释放掉，数据还没来得及缓存的时候，其他线程进来发现缓存中还是没有数据，就又会去查询数据库，造成缓存击穿的问题。
public Map<String, List<Catelog2VO>> getCatelogJsonFromDB() { synchronized (this){ //得到锁以后我们应该再去缓存中查看一次，如果没有才继续确定查询 String catelogJson = stringRedisTemplate.opsForValue().get("catelogJson"); if (!StringUtils.isEmpty(catelogJson)){ //缓存不为空直接返回 Map<String, List<Catelog2VO>> result = JSON.parseObject(catelogJson, new TypeReference<Map<String, List<Catelog2VO>>>() { }); return result; } System.out.println("开始查数据库喽...."); //先将数据库中所有的分类都遍历出来 List<CategoryEntity> selectList = baseMapper.selectList(null); // 一级分类 List<CategoryEntity> level1Categorys = this.getLevel1Categorys(); //封装数据 Map<String, List<Catelog2VO>> map = level1Categorys.stream() .collect(Collectors.toMap(k -> k.getCatId().toString(), v -> { //遍历每一个一级分类，查到一级分类对应的二级分类 List<CategoryEntity> catelog2 = getParentCid(selectList, v.getCatId()); //封装上面的结果 List<Catelog2VO> catelog2VOS = null; if (catelog2 != null) { catelog2VOS = catelog2.stream().map(item2 -> { Catelog2VO catelog2VO = new Catelog2VO(item2.getParentCid().toString(), null, item2.getCatId().toString(), item2.getName()); //找当前二级分类的三级分类封装成vo List<CategoryEntity> catelog3 = getParentCid(selectList, item2.getCatId()); if (catelog3 != null) { List<Catelog2VO.Catelog3VO> catelog3VOS = catelog3.stream().map(item3 -> { Catelog2VO.Catelog3VO catelog3VO = new Catelog2VO.Catelog3VO(item3.getParentCid().toString(), item3.getCatId().toString(), item3.getName()); return catelog3VO; }).collect(Collectors.toList()); catelog2VO.setCatalog3List(catelog3VOS); } return catelog2VO; }).collect(Collectors.toList()); } return catelog2VOS; })); //将查到的数据放入缓存，将对象转为json String valueJson = JSON.toJSONString(map); stringRedisTemplate.opsForValue().set("catelogJson",valueJson,1,TimeUnit.DAYS); return map; } }
不足
本地锁在分布式下的问题：本地锁在分布式的情况下锁不住进程。
分布式锁
原理
我们可以同时去一个地方占坑，如果占到，就去执行逻辑，否则就等待，直到锁的释放。等待锁释放可以使用自旋的方式。
分布式下死锁问题
死锁问题简单来说某一个进程拿到了锁，并开始执行业务的时候突然发生宕机的情况，这就导致锁还没删，其他进程也拿不到锁，线程就会不断等待。
解决方法：给锁设置过期时间，注意此处加入过期时间一定要和加锁是原子性的，不然的话可能在添加过期时间的时候宕机，还是死锁
SET resource_name my_random_value NX PX 30000
分布式下删锁的问题
假如拿到锁的进行业务时间大于锁的自动过期时间，这就会导致当业务还没执行结束锁已经删除，其他进程进来拿到了锁，当前面的进程业务执行完进行删锁的时候删的却是其他进程的锁，这时候又会有其他进程进来了，继续重复这个流程。
解决方法：设置uuid，每个进程只能删除自己的锁，如果自己的锁已经过期了，就不用删了。但是这种方式存在问题是因为网络是有开销的，假如在拿到缓存中的uuid网络进行往返的时候，锁突然过期，其他进程拿到锁设置自己的uuid，那么之前的进程拿到的uuid和自己设置的uuid进行比对，发现相同进行删除缓存的锁，这时候删除的还是别人的锁。
所以最终解决的方案是：判断uuid是否相同和删除锁也要是原子性的。所以可以使用redis的lua脚本进行操作
if redis.call("get",KEYS[1]) == ARGV[1] then return redis.call("del",KEYS[1]) else return 0 end
if (lock){ System.out.println("获取分布式锁成功"); Map<String, List<Catelog2VO>> dataFromDB = null; //加锁成功。。。执行业务 //设置过期时间，必须和加锁是同步的原子的 try { dataFromDB = getCatelogJsonFromDB(); }finally { String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end"; //删除锁 stringRedisTemplate.execute(new DefaultRedisScript<Long>(script,Long.class),Arrays.asList("lock"),uuid); } return dataFromDB; } else { System.out.println("获取分布式锁失败...等待重试..."); //加锁失败，开启重试机制 //休眠一百毫秒 try { TimeUnit.MICROSECONDS.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); } return getCatelogJsonFromDBWithRedisLock(); //自选方式 } }
简单实现
@Override public Map<String, List<Catelog2VO>> getCatelogJson() { // 空结果缓存：解决缓存穿透问题 // 设置过期时间： 解决缓存雪崩 // 加锁：解决缓存击穿问题 //加入缓存逻辑先查看redis中有没有数据，有的话直接返回并且将redis中string数组取出来进行反序列化，没有的调用查数据库的方法进行查询 //查询redis中数据 ValueOperations<String, String> ops = stringRedisTemplate.opsForValue(); String catelogJson = ops.get("catelogJson"); //如果redis中没有数据那么就调方法返回数据然后序列化存到redis中 if (StringUtils.isEmpty(catelogJson)) { //System.out.println("缓存没命中....开始查询数据库...."); Map<String, List<Catelog2VO>> catelogJsonFromDB = getCatelogJsonFromDBWithRedisLock(); // String catelogJsonFromDBString = JSON.toJSONString(catelogJsonFromDB); // ops.set("catelogJson", catelogJsonFromDBString); //取数据 catelogJson = ops.get("catelogJson"); } //将redis中的json数据进行反序列 并返回 return JSON.parseObject(catelogJson, new TypeReference<Map<String, List<Catelog2VO>>>() {}); } public Map<String, List<Catelog2VO>> getCatelogJsonFromDBWithRedisLock() { ValueOperations<String, String> ops = stringRedisTemplate.opsForValue(); String catelogJson = ops.get("catelogJson"); if (!StringUtils.isEmpty(catelogJson)) { //如果此时缓存中有数据则直接返回即可 return JSON.parseObject(catelogJson, new TypeReference<Map<String, List<Catelog2VO>>>() {}); } //将redis中的json数据进行反序列 并返回 //占分布式锁 去redis占坑 String uuid = UUID.randomUUID().toString(); Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", uuid,3000, TimeUnit.MINUTES); if (lock){ System.out.println("获取分布式锁成功"); Map<String, List<Catelog2VO>> dataFromDB = null; //加锁成功。。。执行业务 //设置过期时间，必须和加锁是同步的原子的 try { dataFromDB = getCatelogJsonFromDB(); }finally { String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end"; //删除锁 stringRedisTemplate.execute(new DefaultRedisScript<Long>(script,Long.class),Arrays.asList("lock"),uuid); } return dataFromDB; } else { System.out.println("获取分布式锁失败...等待重试..."); //加锁失败，开启重试机制 //休眠二百毫秒 try { Thread.sleep(200); } catch (InterruptedException e) { e.printStackTrace(); } return getCatelogJsonFromDBWithRedisLock(); //自旋方式 } } public Map<String, List<Catelog2VO>> getCatelogJsonFromDB() { synchronized (this){ //得到锁以后我们应该再去缓存中查看一次，如果没有才继续确定查询 String catelogJson = stringRedisTemplate.opsForValue().get("catelogJson"); if (!StringUtils.isEmpty(catelogJson)){ //缓存不为空直接返回 Map<String, List<Catelog2VO>> result = JSON.parseObject(catelogJson, new TypeReference<Map<String, List<Catelog2VO>>>() { }); return result; } System.out.println("开始查数据库喽...."); //先将数据库中所有的分类都遍历出来 List<CategoryEntity> selectList = baseMapper.selectList(null); // 一级分类 List<CategoryEntity> level1Categorys = this.getLevel1Categorys(); //封装数据 Map<String, List<Catelog2VO>> map = level1Categorys.stream() .collect(Collectors.toMap(k -> k.getCatId().toString(), v -> { //遍历每一个一级分类，查到一级分类对应的二级分类 List<CategoryEntity> catelog2 = getParentCid(selectList, v.getCatId()); //封装上面的结果 List<Catelog2VO> catelog2VOS = null; if (catelog2 != null) { catelog2VOS = catelog2.stream().map(item2 -> { Catelog2VO catelog2VO = new Catelog2VO(item2.getParentCid().toString(), null, item2.getCatId().toString(), item2.getName()); //找当前二级分类的三级分类封装成vo List<CategoryEntity> catelog3 = getParentCid(selectList, item2.getCatId()); if (catelog3 != null) { List<Catelog2VO.Catelog3VO> catelog3VOS = catelog3.stream().map(item3 -> { Catelog2VO.Catelog3VO catelog3VO = new Catelog2VO.Catelog3VO(item3.getParentCid().toString(), item3.getCatId().toString(), item3.getName()); return catelog3VO; }).collect(Collectors.toList()); catelog2VO.setCatalog3List(catelog3VOS); } return catelog2VO; }).collect(Collectors.toList()); } return catelog2VOS; })); //将查到的数据放入缓存，将对象转为json String valueJson = JSON.toJSONString(map); stringRedisTemplate.opsForValue().set("catelogJson",valueJson,1,TimeUnit.DAYS); return map; } }
Redisson
阻塞锁
Redisson可以对redis进行管理，自动完成很多关于缓存的操作，例如加锁
// 1)锁的自动续期。不用担心业务时间长锁自动过期被删除掉，运行期间自动给锁进行续期30秒 // 2)加锁的业务完成后就不会给锁自动续期，即使不解锁也会自动释放
引入依赖
<dependency> <groupId>org.redisson</groupId> <artifactId>redisson</artifactId> <version>3.12.0</version> </dependency>
配置config
@Configuration @ComponentScan @EnableCaching public class MyRedissonConfig { @Bean(destroyMethod="shutdown") RedissonClient redisson() throws IOException { Config config = new Config(); config.useSingleServer() .setAddress("redis://192.168.100.10:6379"); //根据config创建redissonclient实例 RedissonClient redissonClient = Redisson.create(config); return redissonClient; } }
controller
@Autowired private RedissonClient redison; @GetMapping("/hello") public String hello(){ //获取一把锁 只要锁的名字一样 就是同一把锁 RLock lock = redison.getLock("my-lock"); lock.lock(); try { System.out.println("加锁成功等待执行业务....."+ Thread.currentThread().getId()); Thread.sleep(30000); }catch (Exception e){ e.printStackTrace(); }finally { System.out.println("释放锁。。。。"+ Thread.currentThread().getId()); lock.unlock(); } return "hello"; }
读写锁
- 读+ 读 = 无锁
- 读+ 写 = 等读完成才能读
- 写 + 写 = 阻塞状态
- 写 + 读 = 等写完成才能读
@GetMapping("/read") public String read(){ RReadWriteLock readWriteLock = redison.getReadWriteLock("wr-lock"); RLock lock = readWriteLock.readLock(); //读数据 try { lock.lock(30, TimeUnit.SECONDS); ValueOperations<String, String> ops = stringRedisTemplate.opsForValue(); Thread.sleep(10000); String myMsg = ops.get("myMsg"); return myMsg; }catch (Exception e){ e.printStackTrace(); }finally { System.out.println("释放锁。。。。"+ Thread.currentThread().getId()); lock.unlock(); } return "什么也没读到"; } @GetMapping("/write") public String write(){ RReadWriteLock readWriteLock = redison.getReadWriteLock("wr-lock"); RLock lock = readWriteLock.writeLock(); try { lock.lock(30, TimeUnit.SECONDS); ValueOperations<String, String> ops = stringRedisTemplate.opsForValue(); ops.set("myMsg", UUID.randomUUID().toString()); Thread.sleep(10000); }catch (Exception e){ e.printStackTrace(); }finally { System.out.println("释放锁。。。。"+ Thread.currentThread().getId()); lock.unlock(); } return "什么也没读到"; }
闭锁
只有当条件满足后才能释放锁。
//闭锁 @GetMapping("lockDoor") public String door() throws InterruptedException { RCountDownLatch door = redison.getCountDownLatch("door"); door.trySetCount(3); door.await(); return "放假啦！！！！"; } @GetMapping("gogogo/{id}") public String door(@PathVariable Long id) { RCountDownLatch door = redison.getCountDownLatch("door"); door.countDown(); return id + "班走喽！！"; }
信号量
可以设置值作为信号量，当有信号量的时候那么就可以进行操作，没有信号量的时候需要等待其他进程释放后在进行操作。
应用场景例如：限流操作，设置一个应用的最大流量是1w
@GetMapping("setPosition") public String setPosition() throws InterruptedException { RSemaphore park = redison.getSemaphore("park"); park.trySetPermits(3); return "设置了3个车位"; } @GetMapping("park") public String park() throws InterruptedException { RSemaphore park = redison.getSemaphore("park"); park.acquire(); //park.tryAcquire(); 尝试获取信号量，会返回一个布尔值 return "停车成功喽~~~"; } @GetMapping("go") public String go() throws InterruptedException { RSemaphore park = redison.getSemaphore("park"); park.release(); return "车开走喽~~~"; }
