缓存三大问题及解决方案

朴灿烈づ我的快乐病毒、 2021-09-21 05:16 407阅读 0赞

# 1、 为什么使用缓存？ #

随着互联网的普及和发展，用户数和访问量越来越大，我们的应用需要支撑更高的并发量，但是数据库每秒能接受的请求次数也是有限的，这往往成为系统的瓶颈。目前大部分系统都引入了缓存机制从而避免过多的请求直接与数据库交互，极大的提升了系统的qps和稳定性。

# 2、缓存带来的问题以及如何解决 #

引入缓存，比如电商项目常把商品加入缓存redis中，这时查询商品，很多人会不假思索地写出如下类似代码（注：只是举个例子，实际可能不是这样存储）：

public Result<Product> selectProductByCode(String productCode) {
           //查询缓存，valueOperations是spring-data-redis中字符串值操作
           Object redisObj = valueOperations.get(productCode);
    
            //缓存命中
           if(redisObj != null) {
               return new Result().setCode(200).setData(redisObj).setMsg("success");
           }
           Product product = productMapper.selectByProductCode(productCode);
           if (product != null) {
               //将数据库中的数据加入缓存
               valueOperations.set(productCode, product);  
               return new Result().setCode(200).setData(product).setMsg("success");
           }
           return new Result().setCode(500).setData(null).setMsg("查询没有结果");
        }

引入缓存能带来系统性能的提升，但也会带来一些问题：**缓存穿透、缓存击穿、缓存雪崩等**。

## 2.1 缓存穿透 ##

### 描述： ###

缓存穿透是指查询**缓存中没有，数据库中也没有**的数据 。从系统层面来看，请求像是穿透了缓存，直接访问数据库。这种情况会给系统带来潜在的风险，如果有大量的请求访问系统数据库中不存在的数据，会直接请求到数据库层导致DB瘫痪从而引起系统故障，服务不可用。

### 解决方案： ###

**1、缓存空对象**

缓存空对象是在第一次查询到数据库中也不存在的数据后，将该key与对应的空对象也放入缓存中，只不过设定为较短的失效时间，例如几分钟，这样则可以应对短时间的大量的该key攻击，设置为较短的失效时间是因为该值可能业务无关，存在意义不大，且该次的查询也未必是攻击者发起，无过久存储的必要，故可以早点失效。

**缓存空对象代码大致实现：**

首先定义一个空对象：

//定义一个空对象
    public class EmptyObject {
    }

改写上面查询商品selectProductByCode代码：

public Result<Product> selectProductByCode(String productCode) {
           //查询缓存，valueOperations是spring-data-redis中字符串值操作
           Object redisObj = valueOperations.get(productCode);
    
            //缓存命中
           if(redisObj != null) {
               //增加空对象的判断
               if (redisObj instanceof EmptyObject) {
                   return new Result().setCode(500).setData(null).setMsg("查询没有结果");
               }
               return new Result().setCode(200).setData(redisObj).setMsg("success");
           }
           Product product = productMapper.selectByProductCode(productCode);
           if (product != null) {
               //将数据库中的数据加入缓存
               valueOperations.set(productCode, product);  
               return new Result().setCode(200).setData(product).setMsg("success");
           } else {
               //将productCode对应的key的值设为空对象加入到缓存中
               valueOperations.set(product, new EmptyObject(), 5000L, TimeUnit.MILLISECONDS);
           }
           return new Result().setCode(500).setData(null).setMsg("查询没有结果");
        }

**优点**：简单实用

**缺点**：缓存中存储了系统中不存在的数据，占用了缓存的内存

**2、Bloom Filter（布隆过滤器）**

[布隆过滤器][Link 1]（Bloom Filter）是1970年由 Bloom 提出的。它实际上是一个位数组和一系列随机映射函数。布隆过滤器可以用于检索一个元素是否在一个集合中，检索结果是“**可能在集合中**”或“**绝对不在集合中**”。它的优点是空间效率和查询时间都远远超过一般的算法，缺点是有一定的误判率（错误率）和删除处理困难。

![image.png][]

Bloom Filter示意图（图片来自维基百科）

在上图的BloomFilter中，位数组长度为18，有3个Hash函数。其中**x,y,z**在集合中，**w**不在集合中。如果**w**经过三个Hash函数映射到位数组的第3，4，5位上（下标从0开始），那么就会出现误判。**w**并不在集合中，但是第3，4，5位也都是1.

![image.png][image.png 1]

Bloom Filter 过滤示意图

影响Bloom Filter误判率的因素：bit数组的长度和Hash函数的数量。可以更改这两个变量给Bloom Filter设定一个项目系统可以接受的误判率值。

Bloom Filter开源实现可以参考Google guava 中**BloomFilter<T>**的实现：

com.google.common.hash.BloomFilter<T>

分布式系统中Bloom Filter的实现，这里简单介绍一下，缓存使用以常用的redis为例：

**1、可以使用****SETBIT命令构造****Bloom Filter**

基本思想：redis中字符串（SDS）值存储都是按二进制的方式处理的，字符串的API都是二进制安全的，因此可以直接通过SETBIT命令操作字符串转换成二进制后的位数组，这个位数组就可以作为Bloom Filter中的位数组。网上资料很多，此处不详细叙述了。

**2、****RedisBloom模块中提供了Bloom Filter插件**  
    安装了插件后就能使用，详见GitHub地址：[https://github.com/RedisBloom/RedisBloom][https_github.com_RedisBloom_RedisBloom]。

**3、使用Redisson中的RBloomFilter**

**使用布隆过滤器代码解决缓存穿透大致实现：**

上面的写法默认已经根据数据库中的数据初始化了Bloom Filter。

**优点：**占用空间少

**缺点：**1、不支持删除操作。如果要支持删除操作，可以在删除后重新初始化布隆Bloom Filter或者使用改进的Counting Bloom Filter；

2、误判率的存在。

## 2.2 缓存击穿 ##

### 描述： ###

缓存击穿是指查询**缓存中没有，数据库中有**的数据 。缓存击穿是指某个热点key刚好失效，大量的请求都打到数据库上面去了。

### 解决方案： ###

采用分布式的互斥锁。当大量的请求查询该已失效的key时，我们在第一个请求上用互斥锁锁住该key，其他请求就会锁等待。等第一个请求从DB中查询到了数据，并加入到缓存之后，后面的请求就可以直接从缓存中读取数据了。

**大致代码：**

public Result<Product> selectProductByCode(String productCode) {
            //先检索布隆过滤器
            boolean bloomFilterExist = bloomFilter.isExists(productCode);
            if (!bloomFilterExist) {
                return new Result().setCode(500).setData(null).setMsg("查询没有结果");
            }
            //查询缓存
            Object redisObj = valueOperations.get(productCode);
            //缓存命中
            if (redisObj != null) {
                return new Result().setCode(200).setData(redisObj).setMsg("success");
            }
            //分布式锁mutexLock用来锁住key
            mutexLock.lock(productCode);
            try {
                //查询缓存
                redisObj = valueOperations.get(productCode);
                //缓存命中
                if (redisObj != null) {
                    return new Result().setCode(200).setData(redisObj).setMsg("success");
                }
                Product product = productMapper.selectByProductCode(productCode);
                if (product != null) {
                    //将数据库中的数据加入缓存
                    valueOperations.set(productCode, product);
                    return new Result().setCode(200).setData(product).setMsg("success");
                }
                return new Result().setCode(500).setData(null).setMsg("查询没有结果");
            } finally {
                //解锁
                mutexLock.unlock(key);
            }
        }

分布式互斥锁的实现种类很多，基于redis的、zookeeper的等等，可以根据业务场景相应使用。

## 2.3 缓存雪崩 ##

### 描述： ###

缓存雪崩主要有两种情况：一种是缓存机器宕机，服务不可用引起；另一种可能情况是缓存key的集中失效的情况。

### 解决方案： ###

对于缓存机器宕机引起的可以搭建高可用集群Cluster解决；

对于缓存key的集中失效问题可以给key设置不同的过期时间防止集中失效。

[Link 1]: https://en.wikipedia.org/wiki/Bloom_filter
[image.png]: /images/20210920/29dd1e66f1cf49d38b4b7f92e6ab70b6.png
[image.png 1]: /images/20210920/85e8c951f58c47b5aae39da335a2d2af.png
[https_github.com_RedisBloom_RedisBloom]: https://github.com/RedisBloom/RedisBloom