Redis的三大问题-阿里云开发者社区

Redis的三大问题

一般我们对缓存读操作的时候有这么一个固定的套路：

如果我们的数据在缓存里边有，那么就直接取缓存的。
如果缓存里没有我们想要的数据，我们会先去查询数据库，然后将数据库查出来的数据写到缓存中。
最后将数据返回给请求
代码例子:

1 @Override
2 public R selectOrderById(Integer id) {
3 //查询缓存
4 Object redisObj = valueOperations.get(String.valueOf(id));
5
6 //命中缓存
7 if(redisObj != null) {
8 //正常返回数据
9 return new R().setCode(200).setData(redisObj).setMsg("OK");
10 }
11 Order order = orderMapper.selectOrderById(id);
12 if (order != null) {
13 valueOperations.set(String.valueOf(id), order); //加入缓存
14 return new R().setCode(200).setData(order).setMsg("OK");
15 }
16 return new R().setCode(500).setData(new NullValueResultDO()).setMsg("查询无果");
17 }

但这样写的代码是不行的,这代码里就有我们缓存的三大问题的两大问题.穿透,击穿.

一,缓存雪崩
1.1什么是缓存雪崩?
第一种情况:Redis挂掉了,请求全部走数据库.

第二种情况:缓存数据设置的过期时间是相同的,然后刚好这些数据删除了,全部失效了,这个时候全部请求会到数据库

缓存雪崩如果发生了,很有可能会把我们的数据库搞垮,导致整个服务器瘫痪.

1.2如何解决缓存雪崩?
对于第二种情况,非常好解决:

在存缓存的时候给过期时间加上一个随机值,这样大幅度的减少缓存同时过期.

第一种情况:

事发前：实现Redis的高可用(主从架构+Sentinel 或者Redis Cluster)，尽量避免Redis挂掉这种情况发生。
事发中：万一Redis真的挂了，我们可以设置本地缓存(ehcache)+限流(hystrix)，尽量避免我们的数据库被干掉(起码能保证我们的服务还是能正常工作的)
事发后：redis持久化，重启后自动从磁盘上加载数据，快速恢复缓存数据。

二,缓存穿透
2.1什么是缓存穿透?
比如你抢了你同事的女神,你同事很气,想搞你,在你的项目里,每次请求的ID为负数.这个时候缓存肯定是没有的,缓存就没用了,请求就会全部找数据库,但数据库也没用这个值.所以每次返回空出去.

缓存穿透是指查询一个一定不存在的数据。由于缓存不命中，并且出于容错考虑，如果从数据库查不到数据则不写入缓存，这将导致这个不存在的数据每次请求都要到数据库去查询，失去了缓存的意义。

这就是缓存穿透:

请求的数据在缓存大量不命中，导致请求走数据库。

缓存穿透如果发生了，也可能把我们的数据库搞垮，导致整个服务瘫痪！

2.2如何解决缓存穿透?
解决缓存穿透也有两种方案：

由于请求的参数是不合法的(每次都请求不存在的参数)，于是我们可以使用布隆过滤器(BloomFilter)或者压缩filter提前拦截，不合法就不让这个请求到数据库层！
当我们从数据库找不到的时候，我们也将这个空对象设置到缓存里边去。下次再请求的时候，就可以从缓存里边获取了。这种情况我们一般会将空对象设置一个较短的过期时间。
缓存空对象代码例子:

1 public R selectOrderById(Integer id) {
2 return cacheTemplate.redisFindCache(String.valueOf(id), 10, TimeUnit.MINUTES, new CacheLoadble() {
3 @Override
4 public Order load() {
5 return orderMapper.selectOrderById(id);
6 }
7 },false);
8 }

1 public R redisFindCache(String key, long expire, TimeUnit unit, CacheLoadble cacheLoadble, boolean b) {
2 //查询缓存
3 Object redisObj = valueOperations.get(String.valueOf(key));
4 //命中缓存
5 if (redisObj != null) {
6 if(redisObj instanceof NullValueResultDO){
7 return new R().setCode(500).setData(new NullValueResultDO()).setMsg("查询无果");
8 }
9 //正常返回数据
10 return new R().setCode(200).setData(redisObj).setMsg("OK");
11 }
12 try {
13 T load = cacheLoadble.load();//查询数据库
14 if (load != null) {
15 valueOperations.set(key, load, expire, unit); //加入缓存
16 return new R().setCode(200).setData(load).setMsg("OK");
17 }else{
18 valueOperations.set(key,new NullValueResultDO(),expire,unit);
19 }
20
21 } finally {
22
23 }
24 return new R().setCode(500).setData(new NullValueResultDO()).setMsg("查询无果");
25 }

这里封装了一个模板redisFindCache,不然每一个方法都要写这个流程.注意在命中缓存时,要判断数据是否是空对象.

空对象:

1 @Getter
2 @Setter
3 @ToString
4 public class NullValueResultDO{
5
6 }
缓存空对象的缺点:有大量的空数据占用redis的内存.治标不治本.

布隆过滤器:

有谷歌的guava,但是是单机版的,不支持分布式.

也可以用redis的位数组bit手写一个分布式布隆过滤器,代码就不写了.过程就是先把id(比如你是用id为key的)存进布隆过滤器(会经过特定的算法),当我们请求接口的时候先让它查询布隆过滤器,判断数据是否存在.

上面的代码还有个缓存击穿(缓存当中没有,数据库中有)问题,就是并发的时候.比如99个人同时请求,还是会打印99条sql语句,还是会找数据库.

这里的代码是用的分布式锁(互斥锁)

1 public R redisFindCache(String key, long expire, TimeUnit unit, CacheLoadble cacheLoadble,boolean b){
2 //判断是否走过滤器
3 if(b){
4 //先走过滤器
5 boolean bloomExist = bloomFilter.isExist(String.valueOf(key));
6 if(!bloomExist){
7 return new R().setCode(600).setData(null).setMsg("查询无果");
8 }
9 }
10 //查询缓存
11 Object redisObj = valueOperations.get(String.valueOf(key));
12 //命中缓存
13 if(redisObj != null) {
14 //正常返回数据
15 return new R().setCode(200).setData(redisObj).setMsg("OK");
16 }
17 // RLock lock0 = redisson.getLock("{taibai0}:" + key);
18 // RLock lock1 = redisson.getLock("{taibai1}:" + key);
19 // RLock lock2 = redisson.getLock("{taibai2}:" + key);
20 // RedissonMultiLock lock = new RedissonMultiLock(lock0,lock1, lock2);
21 try {
22 redisLock.lock(key);//上锁
23 // lock.lock();
24 //查询缓存
25 redisObj = valueOperations.get(String.valueOf(key));
26 //命中缓存
27 if(redisObj != null) {
28 //正常返回数据
29 return new R().setCode(200).setData(redisObj).setMsg("OK");
30 }
31 T load = cacheLoadble.load();//查询数据库
32 if (load != null) {
33 valueOperations.set(key, load,expire, unit); //加入缓存
34 return new R().setCode(200).setData(load).setMsg("OK");
35 }
36 return new R().setCode(500).setData(new NullValueResultDO()).setMsg("查询无果");
37 }finally {
38 redisLock.unlock(key);//解锁
39 // lock.unlock();
40 }
41 }

三,缓存与数据库双写一致
3.1什么是缓存与数据库双写一致问题?
如果仅仅查询的话，缓存的数据和数据库的数据是没问题的。但是，当我们要更新时候呢？各种情况很可能就造成数据库和缓存的数据不一致了。

这里不一致指的是：数据库的数据跟缓存的数据不一致

从理论上说，只要我们设置了键的过期时间，我们就能保证缓存和数据库的数据最终是一致的。因为只要缓存数据过期了，就会被删除。随后读的时候，因为缓存里没有，就可以查数据库的数据，然后将数据库查出来的数据写入到缓存中。

除了设置过期时间，我们还需要做更多的措施来尽量避免数据库与缓存处于不一致的情况发生。

3.2对于更新操作
一般来说，执行更新操作时，我们会有两种选择：

先操作数据库，再操作缓存
先操作缓存，再操作数据库
首先，要明确的是，无论我们选择哪个，我们都希望这两个操作要么同时成功，要么同时失败。所以，这会演变成一个分布式事务的问题。

所以，如果原子性被破坏了，可能会有以下的情况：

操作数据库成功了，操作缓存失败了。
操作缓存成功了，操作数据库失败了。
如果第一步已经失败了，我们直接返回Exception出去就好了，第二步根本不会执行。

下面我们具体来分析一下吧。

3.2.1操作缓存
操作缓存也有两种方案：

更新缓存
删除缓存
一般我们都是采取删除缓存缓存策略的，原因如下:

高并发环境下，无论是先操作数据库还是后操作数据库而言，如果加上更新缓存，那就更加容易导致数据库与缓存数据不一致问题。(删除缓存直接和简单很多)
如果每次更新了数据库，都要更新缓存【这里指的是频繁更新的场景，这会耗费一定的性能】，倒不如直接删除掉。等再次读取时，缓存里没有，那我到数据库找，在数据库找到再写到缓存里边(体现懒加载)
基于这两点，对于缓存在更新时而言，都是建议执行删除操作！

3.2.2先更新数据库,再删除缓存
正常情况是这样的:

先操作数据库,成功
在删除缓存,也成功
如果原子性被破坏了:

第一步成功(操作数据库)，第二步失败(删除缓存)，会导致数据库里是新数据，而缓存里是旧数据。
如果第一步(操作数据库)就失败了，我们可以直接返回错误(Exception)，不会出现数据不一致。
如果在高并发的场景下，出现数据库与缓存数据不一致的概率特别低，也不是没有：

缓存刚好失效
线程A查询数据库，得一个旧值
线程B将新值写入数据库
线程B删除缓存
线程A将查到的旧值写入缓存
要达成上述情况，还是说一句概率特别低：

因为这个条件需要发生在读缓存时缓存失效，而且并发着有一个写操作。而实际上数据库的写操作会比读操作慢得多，而且还要锁表，而读操作必需在写操作前进入数据库操作，而又要晚于写操作更新缓存，所有的这些条件都具备的概率基本并不大。

对于这种策略，其实是一种设计模式：Cache Aside Pattern

删除缓存失败的解决思路：

将需要删除的key发送到消息队列中
自己消费消息，获得需要删除的key
不断重试删除操作，直到成功
3.2.3先删除缓存,在更新数据库
正常情况是这样的：

先删除缓存，成功；
再更新数据库，也成功；
如果原子性被破坏了：

第一步成功(删除缓存)，第二步失败(更新数据库)，数据库和缓存的数据还是一致的。
如果第一步(删除缓存)就失败了，我们可以直接返回错误(Exception)，数据库和缓存的数据还是一致的。
看起来是很美好，但是我们在并发场景下分析一下，就知道还是有问题的了：

线程A删除了缓存
线程B查询，发现缓存已不存在
线程B去数据库查询得到旧值
线程B将旧值写入缓存
线程A将新值写入数据库
所以也会导致数据库和缓存不一致的问题。

并发下解决数据库与缓存不一致的思路：