1.如何进行缓存的设计

:::tips 缓存中存放的大多都是热点数据，目的就是防止请求可以直接从缓存中获取到数据，而不用访问Mysql
在使用缓存的时候，有一下几个问题，需要时刻注意，缓存穿透，缓存雪崩，缓存击穿。我们要怎样做去解决这些问题？ :::

1.1 缓存穿透

缓存穿透是指查询一个根本不存在的数据，缓存层和存储层都不会命中，通常出于容错的考虑，如果从存储层查不到数据则不写入缓存层。缓存穿透将导致不存在的数据每次请求都要到存储层去查询，失去了缓存保护后端存储的意义。 :::info 举例:
去玩具店（缓存）买冰墩墩，店主说我们店里都没有冰墩墩（根本不存在）。店主建议你你去玩具工厂（存储层）去问问。这时候你就直接跑工厂去问去了。。。 :::

造成缓存穿透的原因：

自身业务代码或者数据出现问题。
一些恶意攻击、爬虫等造成大量空命中。

缓存穿透问题解决方案：
缓存空对象

布隆过滤器

缓存空对象

String get(String key) {
 // 从缓存中获取数据
 String cacheValue = cache.get(key);
 // 缓存为空
 if (StringUtils.isBlank(cacheValue)) {
     // 从存储中获取
     String storageValue = storage.get(key);
     cache.set(key, storageValue);
     // 如果存储数据为空， 需要设置一个过期时间(300秒)
     if (storageValue == null) {
         cache.expire(key, 60 * 5);
     }
     return storageValue;
 } else {
     // 缓存非空
     return cacheValue;
 }
}

布隆过滤器
对于恶意攻击，向服务器请求大量不存在的数据造成的缓存穿透，还可以用布隆过滤器先做一次过滤，对于不存在的数据布隆过滤器一般都能够过滤掉，不让请求再往后端发送。当布隆过滤器说某个值存在时，这个值可能不存在；当它说不存在时，那就肯定不存在。

布隆过滤器就是一个大型的位数组和几个不一样的无偏 hash 函数。所谓无偏就是能够把元素的 hash 值算得比较均匀。

向布隆过滤器中添加 key 时，会使用多个 hash 函数对 key 进行 hash 算得一个整数索引值然后对位数组长度进行取模运算得到一个位置，每个 hash 函数都会算得一个不同的位置。再把位数组的这几个位置都置为 1 就完成了 add 操作。
向布隆过滤器询问 key 是否存在时，跟 add 一样，也会把 hash 的几个位置都算出来，看看位数组中这几个位置是否都为 1，只要有一个位为 0，那么说明布隆过滤器中这个key 不存在。如果都是 1，这并不能说明这个 key 就一定存在，只是极有可能存在，因为这些位被置为 1 可能是因为其它的 key 存在所致。如果这个位数组比较稀疏，这个概率就会很大，如果这个位数组比较拥挤，这个概率就会降低。

这种方法适用于数据命中不高、数据相对固定、实时性低（通常是数据集较大）的应用场景，代码维护较为复杂，但是缓存空间占用很少。

可以用redisson实现布隆过滤器，引入依赖：

<dependency>
   <groupId>org.redisson</groupId>
   <artifactId>redisson</artifactId>
   <version>3.6.5</version>
</dependency>

示例伪代码：

package com.redisson;
import org.redisson.Redisson;
import org.redisson.api.RBloomFilter;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
public class RedissonBloomFilter {
    public static void main(String[] args) {
        Config config = new Config();
        config.useSingleServer().setAddress("redis://localhost:6379");
        //构造Redisson
        RedissonClient redisson = Redisson.create(config);
        RBloomFilter<String> bloomFilter = redisson.getBloomFilter("nameList");
        //初始化布隆过滤器：预计元素为100000000L,误差率为3%,根据这两个参数会计算出底层的bit数组大小
        bloomFilter.tryInit(100000000L,0.03);
        //将zhuge插入到布隆过滤器中
        bloomFilter.add("zhuge");
        //判断下面号码是否在布隆过滤器中
        System.out.println(bloomFilter.contains("guojia"));//false
        System.out.println(bloomFilter.contains("baiqi"));//false
        System.out.println(bloomFilter.contains("zhuge"));//true
    }
}

使用布隆过滤器需要把所有数据提前放入布隆过滤器，并且在增加数据时也要往布隆过滤器里放，布隆过滤器缓存过滤伪代码：

/初始化布隆过滤器
RBloomFilter<String> bloomFilter = redisson.getBloomFilter("nameList");
//初始化布隆过滤器：预计元素为100000000L,误差率为3%
bloomFilter.tryInit(100000000L,0.03);
//把所有数据存入布隆过滤器
void init(){
    for (String key: keys) {
        bloomFilter.put(key);
    }
}
String get(String key) {
    // 从布隆过滤器这一级缓存判断下key是否存在
    Boolean exist = bloomFilter.contains(key);
    if(!exist){
        return "";
    }
    // 从缓存中获取数据
    String cacheValue = cache.get(key);
    // 缓存为空
    if (StringUtils.isBlank(cacheValue)) {
        // 从存储中获取
        String storageValue = storage.get(key);
        cache.set(key, storageValue);
        // 如果存储数据为空， 需要设置一个过期时间(300秒)
        if (storageValue == null) {
            cache.expire(key, 60 * 5);
        }
        return storageValue;
    } else {
        // 缓存非空
        return cacheValue;
    }
}

注意：布隆过滤器不能删除数据，如果要删除得重新初始化数据。

1.2 缓存击穿

大量请求在同一时刻获取一个不存在key.导致请求同时都打到DB，可能会造成数据库瞬间压力过大甚至挂掉。 :::info 举例:
大量顾客去玩具店（缓存）买冰墩墩，没有买到。店主说我们店里冰墩墩卖完了，要不你们去工厂买吧。这时候大家都去去玩具工厂（存储层）去买。太多人去订购导致工厂压力多大 :::

示例伪代码：

String get(String key) {
    // 从缓存中获取数据
    String cacheValue = cache.get(key);
    // 缓存为空
    if (StringUtils.isBlank(cacheValue)) {
        // 从存储中获取
        String storageValue = storage.get(key);
        cache.set(key, storageValue);
        //设置一个过期时间(300到600之间的一个随机数)
        int expireTime = new Random().nextInt(300)  + 300;
        if (storageValue == null) {
            cache.expire(key, expireTime);
        }
        return storageValue;
    } else {
        // 缓存非空
        return cacheValue;
    }
}

1.3 缓存雪崩

由于大批量缓存在同一时间失效可能导致大量请求同时穿透缓存直达数据库(针对大量不同的缓存key)

缓存雪崩指的是缓存层支撑不住或宕掉后，流量会像奔逃的野牛一样，打向后端存储层。
由于缓存层承载着大量请求，有效地保护了存储层，但是如果缓存层由于某些原因不能提供服务(比如超大并发过来，缓存层支撑不住，或者由于缓存设计不好，类似大量请求访问bigkey，导致缓存能支撑的并发急剧下降)，于是大量请求都会打到存储层，存储层的调用量会暴增，造成存储层也会级联宕机的情况。 :::info 举例：
由于冰墩墩太过火爆，一时间玩具店里的冰墩墩被抢光，甚至雪容融都全部被卖完。大量顾客都直接去工厂买冰墩墩，雪容融。最后多家工厂处于瘫痪状态。 :::

解决方式:
预防和解决缓存雪崩问题，可以从以下三个方面进行着手。
对于这种情况我们在批量增加缓存时最好将这一批数据的缓存过期时间设置为一个时间段内的不同时间。
1）保证缓存层服务高可用性，比如使用Redis Sentinel或Redis Cluster。
2）依赖隔离组件为后端限流熔断并降级。比如使用Sentinel或Hystrix限流降级组件。
比如服务降级，我们可以针对不同的数据采取不同的处理方式。当业务应用访问的是非核心数据（例如电商商品属性，用户信息等）时，暂时停止从缓存中查询这些数据，而是直接返回预定义的默认降级信息、空值或是错误提示信息；当业务应用访问的是核心数据（例如电商商品库存）时，仍然允许查询缓存，如果缓存缺失，也可以继续通过数据库读取。
3）提前演练。在项目上线前，演练缓存层宕掉后，应用以及后端的负载情况以及可能出现的问题，在此基础上做一些预案设定。

缓存穿透、缓存击穿、缓存雪崩总结 :::info 缓存中存放的大多都是热点数据，目的就是防止请求可以直接从缓存中获取到数据，而不用访问Mysql。 :::

缓存雪崩：如果缓存中某一时刻大批热点数据同时过期，那么就可能导致大量请求直接访问Mysql了，解决办法就是在过期时间上增加一点随机值，另外如果搭建一个高可用的Redis集群也是防止缓存雪崩的有效手段
缓存击穿：和缓存雪崩类似，缓存雪崩是大批热点数据失效，而缓存击穿是指某一个热点key突然失效，也导致了大量请求直接访问Mysql数据库，这就是缓存击穿，解决方案就是考虑这个热点key不设过期时间
缓存穿透：假如某一时刻访问redis的大量key都在redis中不存在（比如黑客故意伪造一些乱七八糟的key），那么也会给数据造成压力，这就是缓存穿透，解决方案是使用布隆过滤器，它的作用就是如果它认为一个key不存在，那么这个key就肯定不存在，所以可以在缓存之前加一层布隆过滤器来拦截不存在的key

2.热点缓存key重建优化

高并发下重建缓存带来的问题

使用“缓存+过期时间”的策略既可以加速数据读写，又保证数据的定期更新，这种模式基本能够满足绝大部分需求。
但是有两个问题如果同时出现，可能就会对应用造成致命的危害：

当前key是一个热点key（例如一个热门的娱乐新闻），并发量非常大。
重建缓存不能在短时间完成，可能是一个复杂计算，例如复杂的SQL等。在缓存失效的瞬间，有大量线程来重建缓存，造成后端负载加大，甚至可能会让应用崩溃。
解决方案
要解决这个问题主要就是要避免大量线程同时重建缓存。
我们可以利用互斥锁来解决，此方法只允许一个线程重建缓存，其他线程等待重建缓存的线程执行完，重新从缓存获取数据即可。
示例伪代码：
```
String get(String key) {
 // 从Redis中获取数据
 String value = redis.get(key);
 // 如果value为空， 则开始重构缓存
 if (value == null) {
     // 只允许一个线程重建缓存， 使用nx， 并设置过期时间ex
     String mutexKey = "mutext:key:" + key;
     if (redis.set(mutexKey, "1", "ex 180", "nx")) {
          // 从数据源获取数据
         value = db.get(key);
         // 回写Redis， 并设置过期时间
         redis.setex(key, timeout, value);
         // 删除key_mutex
         redis.delete(mutexKey);
     }// 其他线程休息50毫秒后重试
     else {
         Thread.sleep(50);
         get(key);
     }
 }
 return value;
}
```
3.数据库与缓存数据不一致
在大并发的场景下,同时操作数据库与缓存会存在数据不一致的情况，如：双写不一致，读写并发不一致。
3.1 双写不一致
:::info 图书馆借同一类书（假如stock=2），小明借了一本书（此时stock=1）正准备更新书的数量，此时来了一个电话。接电话期间，小红，还了一本，并更新数目（stock=2），此时小明接完电话，更新数量（stock=1）。 :::

3.2 读写并发不一致

:::info t1写数据库（stock=1），并删除了缓存。此时t3过来查询，发现缓存为空，于是查询数据库（stock=1），在准备更新缓存的期间，t2抢到了cpu的执行权，进行写入（stock=2）并删除了缓存，这时候t3恢复执行，更新缓存（stock=1）。
这里就出现问题了。数据库现在是（stock=2），但是缓存中（stock=1）。在下次更新缓存的期间，线程查询的缓存与数据库的数据就不一致。 :::

Redis缓存设计 - 图3
解决方案：

对于并发几率很小的数据(如个人维度的订单数据、用户数据等)，这种几乎不用考虑这个问题，很少会发生缓存不一致，可以给缓存数据加上过期时间，每隔一段时间触发读的主动更新即可。
就算并发很高，如果业务上能容忍短时间的缓存数据不一致(如商品名称，商品分类菜单等)，缓存加上过期时间依然可以解决大部分业务对于缓存的要求。
如果不能容忍缓存数据不一致，可以通过加读写锁保证并发读写或写写的时候按顺序排好队，读读的时候相当于无锁。
也可以用阿里开源的canal通过监听数据库的binlog日志及时的去修改缓存，但是引入了新的中间件，增加了系统的复杂度。

读写锁:
读写锁特点特点：读锁是共享锁，写锁是排他锁，读锁和写锁不能同时存在
插队策略：为了防止线程饥饿，读锁不能插队
升级策略：只能降级，不能升级
ReentrantReadWriteLock适合于读多写少的场合，可以提高并发效率，而ReentrantLock适合普通场合

Redis和Mysql如何保证数据一致

先更新Mysql，再更新Redis，如果更新Redis失败，可能仍然不一致
先删除Redis缓存数据，再更新Mysql，再次查询的时候在将数据添加到缓存中，这种方案能解决1方案的问题，但是在高并发下性能较低，而且仍然会出现数据不一致的问题，比如线程1删除了Redis缓存数据，正在更新Mysql，此时另外一个查询再查询，那么就会把Mysql中老数据又查到Redis中
延时双删，步骤是：先删除Redis缓存数据，再更新Mysql，延迟几百毫秒再删除Redis缓存数据，这样就算在更新Mysql时，有其他线程读了Mysql，把老数据读到了Redis中，那么也会被删除掉，从而把数据保持一致(但是这也是有一定时间的数据不一致)
如果不能容忍缓存数据不一致，可以通过加读写锁保证并发读写或写写的时候按顺序排好队，读读的时候相当于无锁。

Redis缓存设计