前言

在我们日常的开发中，无不都是使用数据库来进行数据的存储，但当遇到大量数据并发请求的需求，如秒杀、热点数据请求等，若所有请求都直接打到数据库上会占用大量的硬盘资源，系统在极短的时间内完成成千上万次的读/写操作，极其容易造成数据库系统瘫痪。

此时我们会引入缓存层来阻挡大部分的请求，减轻数据库压力。但引入缓存层往往带来缓存穿透，缓存击穿，缓存雪崩等问题。

本文以Redis为例模拟且解决以上三个问题。

缓存击穿

缓存击穿是指缓存中没有但数据库中有的数据（一般是缓存时间到期），这时由于并发用户特别多，同时读缓存没读到数据，又同时去数据库去取数据，此时如果你的代码没有实现同步机制，会造成小部分的请求直接打到数据库上，给数据库带来一定的压力。

模拟需求

模拟需求：某秒杀活动即将开始，模拟1w个请求同时发生，要获取某商品的商品详情信息

期望：只能有1个请求打到数据库，其他请求均打到Redis或其他缓存中

错误示例

我们先看错误示例，以下示例代码没有做任何同步，模拟情况一。

 import java.util.HashMap; import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.atomic.AtomicInteger; /** * @author HeyS1 * @date 2020/3/12 * @description */ public class ConcurrentTest { //请求次数 private int reqestQty = 10000; //倒计时器，当发送reqestQty次请求后继续执行主线程 private CountDownLatch latch = new CountDownLatch(reqestQty); //记录请求落在数据库上的次数 private AtomicInteger dbSelectCount = new AtomicInteger(); //记录请求落在缓存中的次数 private AtomicInteger cacheSelectCount = new AtomicInteger(); //用HashMap模拟缓存储存 private Map<String, String> cache = new HashMap<>(); public static void main(String[] args) { new ConcurrentTest().go(); } private void go() { //同时创建1w个线程获取 for (int i = 0; i < reqestQty; i++) { new Thread(() -> { this.getGoodsDetail("商品id"); latch.countDown(); }).start(); } // 计数器大于0 时，await()方法会阻塞程序继续执行 try { latch.await(); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("数据库查询次数：" + dbSelectCount.get()); System.out.println("缓存查询次数：" + cacheSelectCount.get()); } /** * 获取商品数据 * * @param key 商品id * @return */ public String getGoodsDetail(String key) { //先从缓存查询，存在则直接返回 String data = this.selectCache(key); if (data != null) { return data; } //不存在则从数据库查询且将数据放入缓存 data = this.selectDB(key); cache.put(key, data); return data; } /** * 从缓存中获取数据 * * @param key * @return */ public String selectCache(String key) { cacheSelectCount.addAndGet(1);//记录次数 System.out.println(Thread.currentThread().getId() + " 从cache获取数据===="); return cache.get(key); } /** * 从数据库中获取数据 * * @param key * @return */ public String selectDB(String key) { sleep(100);//模拟查询数据库花费100ms dbSelectCount.addAndGet(1);//记录次数 System.out.println(Thread.currentThread().getId() + " 从db获取数据===="); return "数据中的数据"; } private static void sleep(long m) { try { Thread.sleep(m); } catch (InterruptedException e) { e.printStackTrace(); } } }

 

结果： 数据库查询次数：202 缓存查询次数：10000

可以看出，如果getGoodsDetail方法没做任何处理，还是会有少数请求直接打到数据库，这就是缓存穿透

 /** * 获取商品数据 * * @param key 商品id * @return */ public String getGoodsDetail(String key) { //先从缓存查询，存在则直接返回 String data = this.selectCache(key); if (data != null) { return data; } //不存在则从数据库查询且将数据放入缓存 data = this.selectDB(key); cache.put(key, data); return data; }

 

解决方案1：synchronized 

使用synchronized 同步代码块可解决该问题，但此方案只适合单机版架构，不适合集群或分布式架构

只需修改一下上面示例中的getGoodsDetail方法即可

 /** * 获取商品数据 * * @param key 商品id * @return */ public String getGoodsDetail(String key) { //先从缓存查询，存在则直接返回 String data = this.selectCache(key); if (data != null) { return data; } //同步代码块 synchronized (this) { //这里还需要再次查询缓存，防止其他等待进入同步代码块的线程的查询打到数据库上 data = this.selectCache(key); if (data != null) { return data; } //不存在则从数据库查询且将数据放入缓存 data = this.selectDB(key); cache.put(key, data); return data; } } 

数据库查询次数：1 缓存查询次数：10276

解决方案2：redis分布式锁

该方案适合集群或分布式架构，单机使用也可以，但没意义。

分布式锁实现方式有一般有3种，本文使用Redis来实现

1. 数据库乐观锁；
2. 基于Redis的分布式锁；
3. 基于ZooKeeper的分布式锁

首先，为了确保分布式锁可用，我们至少要确保锁的实现同时满足以下四个条件：

1. 互斥性。在任意时刻，只有一个客户端能持有锁。 
2. 不会发生死锁。即使有一个客户端在持有锁的期间崩溃而没有主动解锁，也能保证后续其他客户端能加锁。
3. 具有容错性。只要大部分的Redis节点正常运行，客户端就可以加锁和解锁。 
4. 解铃还须系铃人。加锁和解锁必须是同一个客户端，客户端自己不能把别人加的锁给解了。

 

完整示例：

这里用到RedisTemplate，请自行使用Spring集成Redis

 @RunWith(SpringRunner.class) @SpringBootTest(classes = App.class) @Slf4j public class ConcurrentTest3 { @Autowired RedisTemplate<String, String> redisTemplate; //请求次数 private int reqestQty = 10000; //倒计时器，当发送reqestQty次请求后继续执行主线程 private CountDownLatch latch = new CountDownLatch(reqestQty); //记录请求落在数据库上的次数 private AtomicInteger dbSelectCount = new AtomicInteger(); //记录请求落在缓存中的次数 private AtomicInteger cacheSelectCount = new AtomicInteger(); @Test public void go() { //同时创建1w个线程获取 for (int i = 0; i < reqestQty; i++) { new Thread(() -> { this.getGoodsDetail("商品id"); latch.countDown(); }).start(); } //计数器大于0 时，await()方法会阻塞程序继续执行 try { latch.await(); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("数据库查询次数：" + dbSelectCount.get()); System.out.println("缓存查询次数：" + cacheSelectCount.get()); } public String getGoodsDetail(String key) { //先从缓存查询，存在则直接返回 String data = this.selectCache(key); if (data != null) { return data; } /** * requestId主要用来确保解锁的时候，A客户端不要把B客户端获得的锁给释放了，在本例子中其实不存在这种情况 * 看具体业务情况去取舍即可 */ String lockKey = "锁的Key"; String requestId = "请求客户端ID"; //加锁 if (!this.lock(lockKey, requestId, 10)) { //加锁失败，证明其他线程已获得了锁，此时只需等待一会，再次调用本方法即可 sleep(100); this.getGoodsDetail(key); } //加锁成功 //这里还需要再次查询缓存，防止其他等待的线程获得锁时又打到数据库上 data = this.selectCache(key); if (data != null) { return data; } //从数据库查询且将数据放入缓存 data = this.selectDB(key); redisTemplate.opsForValue().set(key, data, 60, TimeUnit.SECONDS); //释放锁 this.unLock(lockKey, requestId); return data; } /** * 使用redis特性实现互斥锁（setnx） * * @param lockKey * @param requestId * @param expireTime 锁过期时间，即超过该时间仍然未被解锁，则自动解锁，防止死锁 * @return */ public boolean lock(String lockKey, String requestId, int expireTime) { Boolean res = redisTemplate.opsForValue().setIfAbsent(lockKey, requestId, Duration.ofSeconds(expireTime)); return res != null && res; } /** * 释放锁，使用Lua脚本，确保原子性 * * @param lockKey * @param requestId * @return */ public boolean unLock(String lockKey, String requestId) { String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end"; DefaultRedisScript<Boolean> redisScript = new DefaultRedisScript<>(script, Boolean.class); Boolean res = redisTemplate.execute(redisScript, Collections.singletonList(lockKey), requestId); return res != null && res; } /** * 从缓存中获取数据 * * @param key * @return */ public String selectCache(String key) { cacheSelectCount.addAndGet(1);//记录次数 System.out.println(Thread.currentThread().getId() + " 从cache获取数据===="); return redisTemplate.opsForValue().get(key); } /** * 从数据库中获取数据 * * @param key * @return */ public String selectDB(String key) { sleep(100);//模拟查询数据库花费100ms dbSelectCount.addAndGet(1);//记录次数 System.out.println(Thread.currentThread().getId() + " 从db获取数据===="); return "数据中的数据"; } private static void sleep(long m) { try { Thread.sleep(m); } catch (InterruptedException e) { e.printStackTrace(); } }

结果： 数据库查询次数：1 缓存查询次数：32095

参考文章：Redis实现分布式锁的正确使用方式(java版本)

 

解决方案拓展

只要实现了同步机制，基本就可以从根本上解决击穿的问题，当然，我们还有一些方法可以去避免发生穿透的发生，比如

1. 热点数据永不过期；
2. 缓存预热，比如秒杀活动开始前，先在redis初始化数据。
3. 编写脚本，去扫描即将过期但此时访问量巨大的缓存，去延迟它的过期时间。

 

缓存穿透

正常情况下，我们去查询数据都是存在。

那么请求去查询一条压根儿数据库中根本就不存在的数据，也就是缓存和数据库都查询不到这条数据，但是请求每次都会打到数据库上面去。

这种查询不存在数据的现象我们称为缓存穿透。

一般情况是黑客攻击，拿着不存在的ID去发送大量的请求，这样产生的请求到数据库去查询，可能会导致你的数据库由于压力过大而宕掉。

解决方案1：缓存空值

之所以会发生穿透，就是因为缓存中没有存储这些空数据的key。从而导致每次查询都到数据库去了。

那么我们就可以为这些key对应的值设置为null或空字符串 丢到缓存里面去。后面再出现查询这个key 的请求的时候，直接返回null 。

这样，就不用在到数据库中去走一圈了，但是别忘了设置过期时间且时间不宜过长，如5分钟。

解决方案2：布隆过滤器

方案1基本能解决业务场景上的问题，但是遇到网站攻击，不停给redis请求不一样的Key，会导致redis内存爆掉。

此时就需要用到另一种方案：布隆过滤器

布隆过滤器是一个神奇的数据结构，可以用来判断一个元素是否在一个集合中

具体自行查阅网上文章

 

缓存雪崩

缓存雪崩其实是概念性问题，和缓存击穿相似。

缓存雪崩是指缓存中数据大批量到过期时间，而查询数据量巨大，引起数据库压力过大甚至down机。和缓存击穿不同的是:缓存击穿指并发查同一条数据，缓存雪崩是不同数据都过期了，很多数据都查不到从而查数据库。

只要解决了击穿和穿透的问题，就不存在雪崩了，即使某个时间点大量数据过期，也不会直接打到数据库，前提是你的Redis得扛得住。

当然，我们也得尽量避免这个问题，我们可以才用给缓存数据设定随机的过期时间来解决这个问题，避免大量缓存在同一时间过期。

其实上面“解决方案拓展”中提到方法也适用于雪崩，要根据业务灵活运用方案。