缓存击穿

缓存击穿问题也称热点key问题，就是一个高并发访问（该key访问频率高，访问次数多）并且缓存重建业务比较复杂的key突然失效了，大量的请求访问会在瞬间给数据库带来巨大的冲击。

缓存重建业务比较复杂：

缓存在redis数据库中存储，在一定时间后被清除，缓存失效，失效以后需要重新从数据库中查询写入redis，在实际开发中，从数据库中查询并且构建数据并不是查到什么就存储进redis，有些业务比较复杂，需要多表查询的，甚至是要去各种各样的表关联的运算最终得到的结果将其缓存进数据库。这样的业务耗时就比较长，在该时间段内，相当于redis中一直没有缓存，而在这一时间段内，无数请求就无法命中缓存，就直接到达数据库。

常见解决方案：

• 互斥锁

当线程1查询缓存未命中时，去获取互斥锁，获取成功后查询数据库重建缓存数据，在写入缓存之后释放锁。这样做的话，在其他线程来发起请求后，未命中缓存则尝试去获取互斥锁，获取互斥锁失败，则让其进入自旋状态（让线程循环执行抢锁的过程），直到前一个线程释放锁之后，在发起请求，若缓存命中，则返回，若未命中，则获取互斥锁，循环往复。

如图所示：

问题：互相等待，就比如同一时刻中有一千个线程发起请求，但只有一个线程在构建，其他线程都在进行自旋，如果构造时间过久，其他线程只能自旋，而长时间的自旋会让CPU一直在空转，CPU没有办法去执行其他任务，会浪费CPU，性能较差。

• 逻辑过期

可以认为是永不过期，即当下往Redis中存储数据时，不设置过期时间，而是在设置value时添加一个expire字段（在当前时间基础加上一个过期时间），该字段的意义在于提醒我们何时销毁该key，即在逻辑意义上维护的过期时间，而该key在redis中没有过期时间，再加上在redis配置的合适的内存淘汰策略，只要该key写入redis，就一定可以查到，不会出现缓存未命中的情况。

适用情况：在举办活动时添加，活动结束之间将其移除即可。

注意事项：需要判断逻辑时间有无过期

使用详情：

当线程1查询缓存发现逻辑时间已经过期，则尝试获取互斥锁，获取成功，则新开一线程进行查询数据库重建缓存数据，在写入缓存重置逻辑过期时间，最后释放锁，而线程1在此之前将过期数据返回。

在重建缓存期间，如果有新线程发起请求，发现逻辑时间过期，则尝试获取互斥锁，如果获取互斥锁失败，则直接返回过期数据。

如图所示：

方案对比

解决方案	优点	缺点
互斥锁（考虑数据一致性）	没有额外的内存消耗 ，保证一致性，实现简单	线程需要等待，性能受影响，可能有死锁风险
逻辑过期（考虑性能）	线程无需等待，性能较好	不保证一致性，有额外内存消耗，实现复杂

小结：这两种方案都是在解决缓存重建这一段时间内产生的并发问题。

互斥锁：在缓存重建的这段时间内让这些并发的线程串行执行或者相互等待，从而确保安全。确保数据一致性，牺牲了服务的可用性

逻辑过期：在缓存重建这段时间内保证了可用性，牺牲的是数据的一致性（可能访问的是旧数据）。

案例展示：基于互斥锁方式解决缓存击穿问题

需求：修改根据ID查询商铺的业务，基于互斥锁方式来解决缓存击穿问题。

分析业务流程变化

原先业务流程：

修改步骤，在判断缓存是否命中之后，如果未命中，需要先去尝试获取互斥锁，判断是否拿到，如果没有拿到锁，说明已经有线程在更新，不应该继续往下执行，需要休眠一段时间，在重新尝试。如果拿到互斥锁，执行缓存重建，就可以去查询数据库，将查询到的数据写入Redis，随后释放互斥锁，最后返回结果。

新的业务流程如图所示：

注意事项：在该业务流程中使用的所并不是我们平时使用的锁，我们平时使用synchronized或者lock，这种锁的执行逻辑就是拿到锁执行，没拿到锁等待，而该业务流程中的锁的执行逻辑是自定义的，因此要采用自定义的互斥锁（在多个线程并行执行时，只有一条线程成功，其余线程失败），而在学习redis中string类型时中 setnx命令的效果与之相近，当该key不存在的时候存入，如果存在就不存入，这就是一种互斥效果，在大量线程并发访问时，只有一条线程可以成功。因此获取锁为赋值命令，而释放锁则是删除命令（del）。

为了防止程序出故障导致迟迟没有执行删除命令，因此在设置setnx时通常都会为其设置有效期，来防止锁一直不释放，造成死锁，导致业务故障。

这与真正的互斥锁还是有所差距的，但是在这里够用了。

前置代码：

声明两个方法代表获取锁以及释放锁

// 尝试获取锁
private boolean tryLock(String key){//设置有效期时间取决于业务执行时间，一般比业务时间长一些即可。Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", 10, TimeUnit.SECONDS);//建议不要直接返回flag，防止返回空指针，因为Boolean是boolean的包装类，需要进行拆箱操作，可能导致空指针 网络问题或者键不存在但Redis未响应，可能会返回null，因此需要实用工具类判断。改成BooleanUtil.isTrue(flag)。return  BooleanUtil.isTrue(flag);
}
//  释放锁
private void unlock(String key){stringRedisTemplate.delete(key);
}

流程说明：

进行方法封装，在queryWithMutex()方法中进行缓存重建的业务代码。

代码展示：

public Result queryById(Long id) {//缓存穿透//        Shop shop = queryWithPassThrough(id);//互斥锁解决缓存击穿Shop shop = queryWithMutex(id);if (shop == null) {return Result.fail("店铺不存在");}//  7.返回return Result.ok(shop);}//互斥锁解决缓存穿透public Shop queryWithMutex(Long id) {//1.从redis中查询商铺缓存String shopJson = stringRedisTemplate.opsForValue().get(CACHE_SHOP_KEY + id);Shop shop = JSONUtil.toBean(shopJson, Shop.class);// 2.判断是否存在if (StrUtil.isNotBlank(shopJson)) {//  3.存在，直接返回return shop;}//  4.判断命中的是否是空值if (shopJson != null) {//  返回错误信息return null;}//4.实现缓存重建//4.1 获取互斥锁String lock = LOCK_SHOP_KEY + id;try {boolean isLock = tryLock(lock);//4.2.判断是否获取成功if (!isLock) {//4.3.失败，则休眠并重试Thread.sleep(50);return queryWithMutex(id);}//  4.2.获取锁成功，再次检查redis缓存 判断是否为空，如果存在shopJson = stringRedisTemplate.opsForValue().get(CACHE_SHOP_KEY + id);if (StrUtil.isNotBlank(shopJson)) {//  3.存在，直接返回shop = JSONUtil.toBean(shopJson, Shop.class);return shop;}//4.4.成功，根据id查询数据库shop = getById(id);//模拟重建的延时Thread.sleep(200);if (shop == null) {//将空值返回redisstringRedisTemplate.opsForValue().set(CACHE_SHOP_KEY + id, "", CACHE_NULL_TTL, TimeUnit.MINUTES);//  5.不存在，返回错误return null;}//  6.存在，写入redisstringRedisTemplate.opsForValue().set(CACHE_SHOP_KEY + id, JSONUtil.toJsonStr(shop), CACHE_SHOP_TTL, TimeUnit.MINUTES);//  7.释放互斥锁}catch (InterruptedException e) {throw new RuntimeException(e);}finally {unlock(lock);}//8.返回return shop;​}//  逻辑过期解决缓存击穿public Shop queryWithPassThrough(Long id){//1.从redis中查询商铺缓存String shopJson = stringRedisTemplate.opsForValue().get(CACHE_SHOP_KEY+id);// 2.判断是否存在if (StrUtil.isNotBlank(shopJson)){//  3.存在，直接返回Shop shop = JSONUtil.toBean(shopJson, Shop.class);return shop;}//  4.判断命中的是否是空值if (shopJson != null){//  返回错误信息return null;}//  4.不存在，根据id查询数据库Shop shop = getById(id);if (shop == null) {//将空值返回redisstringRedisTemplate.opsForValue().set(CACHE_SHOP_KEY+id, "", CACHE_NULL_TTL, TimeUnit.MINUTES);//  5.不存在，返回错误return null;}//  6.存在，写入redisstringRedisTemplate.opsForValue().set(CACHE_SHOP_KEY+id,JSONUtil.toJsonStr(shop),CACHE_SHOP_TTL, TimeUnit.MINUTES);//  7.返回return shop;​}

借助Jmeter工具进行高并发环境模拟

测试结果如下。

案例展示：基于逻辑过期方式解决缓存击穿问题

需求：修改根据ID查询商铺的业务，基于逻辑过期方式来解决缓存击穿问题。

逻辑过期并不是真的过期，他要求存储数据到redis中的时候，额外的添加一个过期时间的字段。

key本身不需要设置TTL，他的过期时间不由redis控制，而是由业务代码判断是否过期，这样在业务上就会复杂很多。要修改业务流程。

修改详情：首先前端去提交商铺的ID到服务端，服务端在通过ID在Redis中查询缓存，而在逻辑过期方式中，缓存不会出现未命中的情况。

原因：key没有过期时间，一旦key添加在缓存中，就会永久存在，除非活动结束，在人工删除，而像这种热点key一般是参加活动的一些商品，或者是一些其他的东西，会提前加入缓存，并设置逻辑过期时间。

因此，在理论情况下，所有的热点key都会提前添加好，并一直存在，直到活动结束，人工删除。

因此可以不用判断是否命中缓存，如果缓存不存在，则说明该key并不在活动中，所以在流程中象征性的判断一下即可，若未命中就直接返回空。

而核心逻辑就在于默认命中之后，在命中后需要判断是否过期，也就是逻辑过期时间。

如果未过期，就直接返回即可。

如果数据过期，说明需要重新加载，需要去做缓存重建。

但是也不能让所有线程都去重建，因此还是需要争抢，即先尝试获取互斥锁，然后判断是否获取到，如果获取失败，说明之前已经有线程在进行更新缓存，这时可以直接返回旧数据。

如果抢夺成功，就需要去执行缓存重建，而且并不是在本线程执行，而是新建线程去执行缓存重建，而本线程先返回旧数据，由该独立线程执行数据重建，查询数据库，将数据写入缓存，并且设置逻辑过期时间，再去释放锁即可。

流程如图所示：

代码实现：

考虑问题：

将数据写入redis的时候，我们要设置一个逻辑过期时间，那逻辑过期时间如何添加数据里？

可以直接找到实体类，在实体类中添加逻辑过期时间字段，但这种方案并不友好，因为对原来的代码和业务逻辑进行了修改，这里有两个方案，

一：在工具类中新建对象RedisData，在该类中新建LocalDatatime类型的字段expireTime，然后让实体类shop集继承redisData类，

二：在RedisData中去添加Object类型的Data，也就是说redisData自己带有过期时间并且里面带有数据，这个数据就是要存入redis的热点key，是一个万能的存储对象。第二种方案不会对原来的代码有任何的修改。

在这里选择第二种方案。而像这种热点key的数据需要提前导入进去，在实际开发中，可能会有一个后台管理系统，可以将某一些热点的数据在后台提前的添加缓存中,但是该项目没有，因此只能基于单元测试的方式，将店铺数据加入缓存中，相当于做一个缓存的预设。

代码展示：

向Redis存入热点数据以及设置逻辑过期时间

 private void saveShop2Redis(Long id, Long expireSeconds) {//  1.查询店铺数据Shop shop = getById(id);//  2.封装逻辑过期时间RedisData redisData = new RedisData();redisData.setData(shop);// 设置逻辑过期时间 现在时间加上过期时间redisData.setExpireTime(LocalDateTime.now().plusSeconds(expireSeconds));//  3.写入redisstringRedisTemplate.opsForValue().set(CACHE_SHOP_KEY+id,JSONUtil.toJsonStr(redisData));}

编写单元测试：

 @Resourceprivate ShopServiceImpl shopService;@Testvoid testSaveShop() {shopService.saveShop2Redis(1L, 10L);}

运行，测试通过，查看redis数据库

数据预热完成

开始解决缓存击穿问题

代码展示：

 public Shop queryWithLogicalExpire(Long id){//1.从redis中查询商铺缓存String shopJson = stringRedisTemplate.opsForValue().get(CACHE_SHOP_KEY+id);// 2.判断是否存在if (StrUtil.isBlank(shopJson)){//  3.不存在，直接返回空return null;}//4.命中，需要把json反序列化为对象RedisData redisData = JSONUtil.toBean(shopJson, RedisData.class);//  4.存在，判断缓存是否过期//Data实际上是jsonObject对象Shop shop = JSONUtil.toBean((JSONObject) redisData.getData(), Shop.class);LocalDateTime expireTime = redisData.getExpireTime();​//  5.判断是否过期if(expireTime.isAfter(LocalDateTime.now())){//  5.1.未过期，直接返回店铺信息return shop;}//  5.2.过期，需要缓存重建//  6.重建缓存// 6.1.获取互斥锁String lock = LOCK_SHOP_KEY + id;boolean isLock = tryLock(lock);// 6.2.判断是否获取锁成功if (isLock) {//再次检测redis缓存是否过期redisData = JSONUtil.toBean(stringRedisTemplate.opsForValue().get(CACHE_SHOP_KEY +     id), RedisData.class);if (redisData.getExpireTime().isAfter(LocalDateTime.now())) {//  6.1.未过期，直接返回。return JSONUtil.toBean((JSONObject) redisData.getData(), Shop.class);}//  6.3.成功，开启独立线程，实现缓存重建CACHE_REBUILD_EXECUTOR.submit(() -> {try {//  重建缓存this.saveShop2Redis(id, 20L);} catch (Exception e) {throw new RuntimeException(e);}finally {//  释放锁unlock(lock);}});}

注意事项：在开启新线程时建议使用线程池，新建线程经常要创建与销毁，十分浪费性能，使用线程池可以做到线程复用。

因此新建线程池执行器。

private static final ExecutorService CACHE_REBUILD_EXECUTOR = Executors.newFixedThreadPool(10);

为了模拟缓存有一定的延迟，可以在数据存入时休眠200ms

 public void saveShop2Redis(Long id, Long expireSeconds) throws InterruptedException {//  1.查询店铺数据Shop shop = getById(id);//  模拟重建的延时Thread.sleep(200);//  2.封装逻辑过期时间RedisData redisData = new RedisData();redisData.setData(shop);// 设置逻辑过期时间 现在时间加上过期时间redisData.setExpireTime(LocalDateTime.now().plusSeconds(expireSeconds));//  3.写入redisstringRedisTemplate.opsForValue().set(CACHE_SHOP_KEY+id,JSONUtil.toJsonStr(redisData));}

延迟越长，越容易出现这种线程安全问题。

开始测试，在高并发的情况下，会不会出现大量线程重建的情况（并发的安全问题），以及一致性问题，在缓存重建完成之前查询到的是旧的数据还是新的数据（将数据库中的数据修改一下进行对比）

新建100个线程并发执行。

检查成果：

测试成功，在前一半线程中，数据为旧数据，后一半线程数据为新数据。

而在数据库只进行了一次数据查询，证明并发安全，只会有一次重建，但是数据一致性会有一些问题。

拓展知识：线程池的具体流程以及如何实现线程复用

线程核心类是ThreadPoolExecutor,它也有很多基于threadpoolexecutor包装的类， ThreadPoolExecutor的七大参数

• corePoolSize (核心线程)：来一个任务开一个线程，直到达到我们的corePoorSize,如果到了这个线程，就不能去开新的线程，将其放在队列中，即workQueue(阻塞队列)

• workQueue（阻塞队列／任务队列）:当核心线程达到以后，之后的任务就放在阻塞队列中 。

• maximumPoolSize（最大线程数）： 当阻塞队列满了以后，需要求其他的线程帮忙，所以会去开额外的线程，但是，corePoolsize加上来帮忙的线程不能超过这个最大线程数。

• keepAliveTime（线程保持活跃时间):代表任务消费完之后，刚刚帮忙的线程可能要回收，也可以设置corePoolSize的线程要回收，KeepAliveTime就是设置要回收的时间，过多久以后回收线程。

• TimeUnit：时间单位

• ThreadFactory（线程工厂）：线程池的核心是线程复用，要去创建线程，这个工厂就代表你如何创建线程，以及你在创建线程他的一些属性设置。

• RejectedExecutionHandler（接口）（拒绝策略），当线程数满了以后，包括开始开启的线程，，包括后面帮你的线程 还有任务队列也满了的时候，不清楚该怎么做，就提供这个接口给你，只要实现RejectedExecution方法就可以了。

执行流程

在初始化线程以后，执行execute方法，execute方法传的是一个任务task（runnable），这个任务是抽象的（接口）. 源码解析：

  //ctl:是一个原子类型，用于保存当前线程池的线程数以及线程状态。//有三位是来保证状态的，还有二十九位用来保存线程数。private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));private static final int COUNT_BITS = Integer.SIZE - 3;private static final int COUNT_MASK = (1 << COUNT_BITS) - 1;​​int c = ctl.get(); //首先，先拿到当前线程池的线程数 if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) //如果说小于corePoolSize（核心线程)，则调用addworker方法。（添加线程来帮助执行任务），也就是说，只要线程数小于核心线程，就会添加一个线程。return;//直接返回c = ctl.get();}if (isRunning(c) && workQueue.offer(command)) {//当大于corePoolSize时，则将添加到阻塞队列中（workqueue），如果if条件成立，则说明阻塞队列添加成功int recheck = ctl.get();if (! isRunning(recheck) && remove(command))reject(command);else if (workerCountOf(recheck) == 0)addWorker(null, false);}else if (!addWorker(command, false)) // 当我的线程数大于corePoolSize并且阻塞队列也是满的，则在调用addworker方法，增加线程，不过此次增加线程的参数为false，false的意思是这个添加的线程是非核心线程，相当于额外的线程reject(command);//如果额外的线程也没有添加成功，就直接拒绝，调用RejectedExecution方法即可（拒绝策略）}

细致解析addworker方法

private boolean addWorker(Runnable firstTask, boolean core) {retry:for (int c = ctl.get();;) {// Check if queue empty only if necessary.if (runStateAtLeast(c, SHUTDOWN)&& (runStateAtLeast(c, STOP)|| firstTask != null|| workQueue.isEmpty()))return false;​for (;;) {//   三元运算符 如果传入的core为true，说明要添加核心线程，那么就去和core对比，如果不为true，则和maximumPoolSize（最大线程数）对比if (workerCountOf(c)>= ((core ? corePoolSize : maximumPoolSize) & COUNT_MASK))return false;if (compareAndIncrementWorkerCount(c))break retry;c = ctl.get();  // Re-read ctlif (runStateAtLeast(c, SHUTDOWN))continue retry;// else CAS failed due to workerCount change; retry inner loop}}//以上代码用于线程判断，判断线程是否关闭，以及线程状态是否正常​boolean workerStarted = false;boolean workerAdded = false;Worker w = null;try {w = new Worker(firstTask); // 开启线程，去找workerfinal Thread t = w.thread; // 开启线程if (t != null) {final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {// Recheck while holding lock.// Back out on ThreadFactory failure or if// shut down before lock acquired.int c = ctl.get();​if (isRunning(c) ||(runStateLessThan(c, STOP) && firstTask == null)) {if (t.getState() != Thread.State.NEW)throw new IllegalThreadStateException();workers.add(w);workerAdded = true;int s = workers.size();if (s > largestPoolSize)largestPoolSize = s;}} finally {mainLock.unlock();}if (workerAdded) {container.start(t);//启动线程workerStarted = true;}}} finally {if (! workerStarted)addWorkerFailed(w);}return workerStarted;}

找到worker构造器

 Worker(Runnable firstTask) {setState(-1); // inhibit interrupts until runWorkerthis.firstTask = firstTask;this.thread = getThreadFactory().newThread(this); //使用线程工厂来创建线程}public void run() {runWorker(this);// 调用run方法}

runworker方法执行任务

final void runWorker(Worker w) {Thread wt = Thread.currentThread();Runnable task = w.firstTask; // task 是我们传的参数 w.firstTask = null;w.unlock(); // allow interruptsboolean completedAbruptly = true;try {while (task != null || (task = getTask()) != null) {// 除了传进来的的以外，我们还要去阻塞队列中去拿，通过getTask（）方法w.lock();// If pool is stopping, ensure thread is interrupted;// if not, ensure thread is not interrupted.  This// requires a recheck in second case to deal with// shutdownNow race while clearing interruptif ((runStateAtLeast(ctl.get(), STOP) ||(Thread.interrupted() &&runStateAtLeast(ctl.get(), STOP))) &&!wt.isInterrupted())wt.interrupt();try {beforeExecute(wt, task);try {task.run(); //执行我们的任务afterExecute(task, null);} catch (Throwable ex) {afterExecute(task, ex);throw ex;}} finally {task = null;w.completedTasks++;w.unlock();}}completedAbruptly = false;} finally {processWorkerExit(w, completedAbruptly);}}

getTask（）方法

private Runnable getTask() {boolean timedOut = false; // Did the last poll() time out?​for (;;) {int c = ctl.get();​// Check if queue empty only if necessary.if (runStateAtLeast(c, SHUTDOWN)&& (runStateAtLeast(c, STOP) || workQueue.isEmpty())) {decrementWorkerCount();return null;}​int wc = workerCountOf(c);​// Are workers subject to culling?boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; //当允许核心线程回收的时候或线程数大于核心线程数，这是可以回收线程， 怎么回收？run方法结束，线程就回收了，run方法什么时候结束？​if ((wc > maximumPoolSize || (timed && timedOut))&& (wc > 1 || workQueue.isEmpty())) {if (compareAndDecrementWorkerCount(c))return null;continue;}​try {//当到达超时时间以后，允许核心线程回收时，就可以回收，如果不允许线程回收或者现在线程数小于核心线程数，就调用take（）方法（一直阻塞）。这也是线程池做到线程复用的关键地方，线程池中的线程不会回收，只会通过阻塞队列阻塞Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take();// 怎么拿到的？take（）方法或poll方法；poll（）方法是超时阻塞，poll是一直阻塞，如果没有任务，要么就是超时阻塞，要么就是一直阻塞，如果有任务就去拿任务if (r != null)return r;timedOut = true;} catch (InterruptedException retry) {timedOut = false;}}}

以上就是源码中的关键源码。以及相应的解释

具体流程

在初始化线程以后，执行execute方法，execute方法传的是一个任务task。

当执行task时，首先判断是否大于corePoolSize（核心线程），如果是，直接丢给workqueue（阻塞队列）。

如果不是，就通过work类中的ThreadFactory（线程工厂）开启线程，执行start方法去Thread中回调run方法。

如果workqueue（阻塞队列）也满了之后，会再次经过一次判断，如果此时线程数小于maximumPoolSize（最大线程数），则再次通过work类中的ThreadFactory（线程工厂）去开启线程。

如果大于maximumPoolSize（最大线程数），则是拒绝策略（RejectedExecutorHandler接口中的RejectedExecution（）方法），

此时run方法有两种选择，当前有task时，执行当前拿到的task，如果没有，自旋去workqueue（阻塞队列）中，会去执行take方法或者poll方法。

take方法是无限制的阻塞，而poll方法则是有超时时间的限制的阻塞，超时时间也可以控制。

当workqueue为空，且允许关闭核心线程 或者当前的线程数大于核心线程数boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;，可以进行线程回收（调用poll方法）。

否则调用take方法，阻塞在take方法中，等待workqueue里面存在task。

这样的话，线程就一直不会被回收，只要代码中 allowCoreThreadTimeOut为false且线程数刚好小于核心线程数

如图所示：