缓存更新和同步策略、缓存穿透、缓存击穿、缓存雪崩-不过是从头再来罢了

1.简介

缓存就是数据交换的缓冲区（称作Cache），是存贮数据的临时地方，一般读写性能较高

实际开发中,会构筑多级缓存来使系统运行速度进一步提升,例如:本地缓存与redis中的缓存并发使用

2.缓存更新策略

	内存淘汰	超时剔除	主动更新
说明	利用Redis的内存淘汰机制，当内存不足时自动淘汰部分数据。下次查询时更新缓存。	给缓存数据添加TTL时间，到期后自动删除缓存。下次查询时更新缓存。	编写业务逻辑，在修改数据库的同时，更新缓存。
一致性	差	一般	好
维护成本	低	超时剔除	高

业务场景：

低一致性需求：使用Redis的内存淘汰机制。例如店铺的查询缓存
高一致性需求：主动更新，并以超时剔除作为兜底方案

2.1 主动更新策略

由于引入了缓存，那么在数据更新时，不仅要更新数据库，而且要更新缓存，这两个更新操作存在前后的问题：

先更新数据库，再更新缓存；
先更新缓存，再更新数据库；

2.1.1 先更新数据库，再更新缓存

举个例子，比如「请求 A 」和「请求 B 」两个请求，同时更新「同一条」数据，则可能出现这样的顺序：

A 请求先将数据库的数据更新为 1，然后在更新缓存前，请求 B 将数据库的数据更新为 2，紧接着也把缓存更新为 2，然后 A 请求更新缓存为 1。

此时，数据库中的数据是 2，而缓存中的数据却是 1，出现了缓存和数据库中的数据不一致的现象。

可能有些同学很好奇，怎么想到这种情况的。我来解答一下，因为我们想模拟的是缓存和数据库中的数据不一致的情况，而更新请求与数据库更新、缓存更新都有关联，所以必然是更新请求，而且，要发生不一致的情况，只能是两个更新请求过程中有交集的情况，否则按顺序访问相安无事

2.1.2 先更新缓存，再更新数据库

依然还是存在并发的问题，分析思路也是一样。假设「请求 A 」和「请求 B 」两个请求，同时更新「同一条」数据，则可能出现这样的顺序：

A 请求先将缓存的数据更新为 1，然后在更新数据库前，B 请求来了，将缓存的数据更新为 2，紧接着把数据库更新为 2，然后 A 请求将数据库的数据更新为 1。

此时，数据库中的数据是 1，而缓存中的数据却是 2，出现了缓存和数据库中的数据不一致的现象。

无论是「先更新数据库，再更新缓存」，还是「先更新缓存，再更新数据库」，这两个方案都存在并发问题，当两个请求并发更新同一条数据的时候，可能会出现缓存和数据库中的数据不一致的现象。

所以此时我们需要考虑另外一种方案：删除缓存。

2.2 Cache Aside Pattern（旁路缓存模式）

Cache Aside Pattern 是我们平时使用比较多的一个缓存读写模式，比较适合读请求比较多的场景。

该策略又可以细分为「读策略」和「写策略」。

写策略的步骤：

更新数据库中的数据；
删除缓存中的数据。

读策略的步骤：

如果读取的数据命中了缓存，则直接返回数据；
如果读取的数据没有命中缓存，则从数据库中读取数据，然后将数据写入到缓存，并且返回给用户。

可能有人问了，在写数据的过程中，不可以先删除 cache ，后更新 db 么？我们来分别模拟一下两种情况

2.2.1 先删除缓存，再更新数据库

假设某个用户的年龄是 20，请求 A 要更新用户年龄为 21，所以它会删除缓存中的内容。这时，另一个请求 B 要读取这个用户的年龄，它查询缓存发现未命中后，会从数据库中读取到年龄为 20，并且写入到缓存中，然后请求 A 继续更改数据库，将用户的年龄更新为 21。

最终，该用户年龄在缓存中是 20（旧值），在数据库中是 21（新值），缓存和数据库的数据不一致。

可以看到，先删除缓存，再更新数据库，在「读 + 写」并发的时候，还是会出现缓存和数据库的数据不一致的问题。

2.2.2 延迟双删

针对「先删除缓存，再更新数据库」方案在「读 + 写」并发请求而造成缓存不一致的解决办法是「延迟双删」

延迟双删实现的伪代码如下：

# 删除缓存
redis.delKey(X)
#更新数据库
db.update(X)
# 睡眠
Thread.sleep(N)
# 再删除缓存
redis.delKey(X)

加了个睡眠时间，主要是为了确保请求 A 在睡眠的时候，请求 B 能够在这这一段时间完成「从数据库读取数据，再把缺失的缓存写入缓存」的操作，然后请求 A 睡眠完，再删除缓存。

所以，请求 A 的睡眠时间就需要大于请求 B 「从数据库读取数据 + 写入缓存」的时间。

但是具体睡眠多久其实是个玄学，很难评估出来，特别是在分布式和高并发场景下。所以这个方案也只是尽可能保证一致性而已，极端情况下，依然也会出现缓存不一致的现象。

因此，用「先更新数据库，再删除缓存」方案的还是较多。

2.2.3 先更新数据库，再删除缓存

继续用「读 + 写」请求的并发的场景来分析。

假如某个用户数据在缓存中不存在，请求 A 读取数据时从数据库中查询到年龄为 20，在未写入缓存中时另一个请求 B 更新数据。它更新数据库中的年龄为 21，并且清空缓存。这时请求 A 把从数据库中读到的年龄为 20 的数据写入到缓存中。

最终，该用户年龄在缓存中是 20（旧值），在数据库中是 21（新值），缓存和数据库数据不一致。

但是在实际中，这个问题出现的概率并不高。因为缓存的写入通常要远远快于数据库的写入，所以在实际中很难出现请求 B 已经更新了数据库并且删除了缓存，请求 A 才更新完缓存的情况。

而一旦请求 A 早于请求 B 删除缓存之前更新了缓存，那么接下来的请求就会因为缓存不命中而从数据库中重新读取数据，所以不会出现这种不一致的情况。

缺点：

「先更新数据库，再删除缓存」的方案虽然保证了数据库与缓存的数据一致性，但是每次更新数据的时候，缓存的数据都会被删除，这样会对缓存的命中率带来影响。

3.缓存同步策略

如何保证数据库与缓存操作都能成功执行

前面我们分析到，无论是更新缓存还是删除缓存，只要第二步发生失败，那么就会导致数据库和缓存不一致。保证第二步成功执行，就是解决问题的关键。我们可以考虑2种解决办法

3.1 重试机制

我们可以引入消息队列，将第二个操作（删除缓存）要操作的数据加入到消息队列，由消费者来操作数据。

如果应用删除缓存失败，可以从消息队列中重新读取数据，然后再次删除缓存，这个就是重试机制。当然，如果重试超过的一定次数，还是没有成功，我们就需要向业务层发送报错信息了。
如果删除缓存成功，就要把数据从消息队列中移除，避免重复操作，否则就继续重试。

3.2 订阅 MySQL binlog，再操作缓存

「先更新数据库，再删缓存」的策略的第一步是更新数据库，那么更新数据库成功，就会产生一条变更日志，记录在 binlog 里。

于是我们就可以通过订阅 binlog 日志，拿到具体要操作的数据，然后再执行缓存删除，阿里巴巴开源的 Canal 中间件就是基于这个实现的。

Canal 模拟 MySQL 主从复制的交互协议，把自己伪装成一个 MySQL 的从节点，向 MySQL 主节点发送 dump 请求，MySQL 收到请求后，就会开始推送 Binlog 给 Canal，Canal 解析 Binlog 字节流之后，转换为便于读取的结构化数据，供下游程序订阅使用。

3.3 如何做到强一致性

前面说过，「先更新数据库，再删除缓存」的方案会对缓存的命中率带来影响。如果我们的业务对缓存命中率有很高的要求，我们可以更新数据库 + 更新缓存 的方案，因为更新缓存并不会出现缓存未命中的情况。

但是这个方案前面我们也分析过，在两个更新请求并发执行的时候，会出现数据不一致的问题，因为更新数据库和更新缓存这两个操作是独立的，而我们又没有对操作做任何并发控制，那么当两个线程并发更新它们的话，就会因为写入顺序的不同造成数据的不一致。

所以我们得增加一些手段来解决这个问题，在更新缓存前先加个分布式锁，保证同一时间只运行一个请求更新缓存，就会不会产生并发问题了，当然引入了锁后，对于写入的性能就会带来影响。

4.三大热点问题

4.1 缓存穿透

缓存穿透是指客户端请求的数据在缓存中和数据库中都不存在，这些请求都会打到数据库。如果有很多恶意的类似请求会给数据库带来很大的压力。

4.1.1 缓存空对象

优点：实现简单，维护方便

缺点：额外的内存消耗(即缓存过多的空对象，可以通过设置过期时间TTL缓解)

如果用 Java 代码展示的话，差不多是下面这样的：

public Object getObjectInclNullById(Integer id) {
    // 从缓存中获取数据
    Object cacheValue = cache.get(id);
    if(cacheValue != null){
        // 之前缓存的空对象
        if(Constant.emptyStr.equals(cacheValue)){
          return new Object();
        }
        //序列化
        Object o = JsonUtils.toJSON(cacheValue,XXX.class);
        retrun o;
    }
    // 从数据库中获取
    Object storageValue = storage.get(key);
    // 如果存储数据为空，需要设置一个过期时间(300秒)
    if (storageValue == null) {
       // 缓存空对象,必须设置过期时间，否则有被攻击的风险
       cache.set(key, Constant.emptyStr,60 * 5,TimeUnit.SECONDS);
       return new Object();
    }
    cache.set(key, storageValue,60 * 60 + randomNum,TimeUnit.SECONDS);
    return storageValue;
}

4.1.2 布隆过滤（BloomFilter）

布隆过滤器是一个非常神奇的数据结构，通过它我们可以非常方便地判断一个给定数据是否存在于海量数据中。

我们可以在写入数据库数据时，使用布隆过滤器做个标记，然后在用户请求到来时，业务线程确认缓存失效后，可以通过查询布隆过滤器快速判断数据是否存在，如果不存在，就不用通过查询数据库来判断数据是否存在。

加入布隆过滤器之后的缓存处理流程图如下。

布隆过滤器由「初始值都为 0 的位图数组」和「 N 个哈希函数」两部分组成。当我们在写入数据库数据时，在布隆过滤器里做个标记，这样下次查询数据是否在数据库时，只需要查询布隆过滤器，如果查询到数据没有被标记，说明不在数据库中。

布隆过滤器会通过 3 个操作完成标记：

第一步，使用 N 个哈希函数分别对数据做哈希计算，得到 N 个哈希值；
第二步，将第一步得到的 N 个哈希值对位图数组的长度取模，得到每个哈希值在位图数组的对应位置。
第三步，将每个哈希值在位图数组的对应位置的值设置为 1；

举个例子，假设有一个位图数组长度为 8，哈希函数 3 个的布隆过滤器。

在数据库写入数据 x 后，把数据 x 标记在布隆过滤器时，数据 x 会被 3 个哈希函数分别计算出 3 个哈希值，然后在对这 3 个哈希值对 8 取模，假设取模的结果为 1、4、6，然后把位图数组的第 1、4、6 位置的值设置为 1。当应用要查询数据 x 是否数据库时，通过布隆过滤器只要查到位图数组的第 1、4、6 位置的值是否全为 1，只要有一个为 0，就认为数据 x 不在数据库中。

布隆过滤器不能直接支持删除工作，因为在删除一个元素时，可能会影响其他元素。有些人可能想到把位数组变成整数数组，每插入一个元素相应的计数器加 1, 这样删除元素时将计数器减掉就可以了。然而要保证安全地删除元素并非如此简单。首先我们必须保证删除的元素的确在布隆过滤器里面。这一点单凭这个过滤器是无法保证的。另外计数器回绕也会造成问题。

布隆过滤器优缺点

优点：

增加和查询元素的时间复杂度为：O(K)(K为哈希函数的个数，一般比较小)，与数据量大小无关
哈希函数相互之间没有关系，方便硬件并行运算
布隆过滤器不需要存储元素本身，在某些对保密要求比较严格的场合有很大优势
在能够承受一定的误判时，布隆过滤器比其他数据结构有这很大的空间优势
数据量很大时，布隆过滤器可以表示全集，其他数据结构不能

缺点：

有误判率，即不能准确判断元素是否在集合中；并且随着存入的元素数量增加，误算率随之增加
不能获取元素本身
一般情况下不能从布隆过滤器中删除元素

布隆过滤器（BloomFilter）使用场景

判断给定数据是否存在：比如判断一个数字是否存在于包含大量数字的数字集中（数字集很大，上亿）、防止缓存穿透（判断请求的数据是否有效避免直接绕过缓存请求数据库）等等、邮箱的垃圾邮件过滤（判断一个邮件地址是否在垃圾邮件列表中）、黑名单功能（判断一个IP地址或手机号码是否在黑名单中）等等。
去重：比如爬给定网址的时候对已经爬取过的 URL 去重、对巨量的 QQ号/订单号去重。

去重场景也需要用到判断给定数据是否存在，因此布隆过滤器主要是为了解决海量数据的存在性问题。

Java实现布隆过滤器

import java.util.BitSet;

/**
 * <p> @Title MyBloomFilter
 * <p> @Description 布隆过滤器实现
 */
public class MyBloomFilter {

    /**
     * 位数组大小
     */
    private static final int DEFAULT_SIZE = 2 << 24;

    /**
     * 通过这个数组创建多个Hash函数
     */
    private static final int[] SEEDS = new int[]{6, 18, 64, 89, 126, 189, 223};

    /**
     * 初始化位数组，数组中的元素只能是 0 或者 1
     */
    private BitSet bits = new BitSet(DEFAULT_SIZE);

    /**
     * Hash函数数组
     */
    private MyHash[] myHashes = new MyHash[SEEDS.length];

    /**
     * 初始化多个包含 Hash 函数的类数组，每个类中的 Hash 函数都不一样
     */
    public MyBloomFilter() {
        // 初始化多个不同的 Hash 函数
        for (int i = 0; i < SEEDS.length; i++) {
            myHashes[i] = new MyHash(DEFAULT_SIZE, SEEDS[i]);
        }
    }

    /**
     * 添加元素到位数组
     */
    public void add(Object value) {
        for (MyHash myHash : myHashes) {
            bits.set(myHash.hash(value), true);
        }
    }

    /**
     * 判断指定元素是否存在于位数组
     */
    public boolean contains(Object value) {
        boolean result = true;
        for (MyHash myHash : myHashes) {
            result = result && bits.get(myHash.hash(value));
        }
        return result;
    }

    /**
     * 自定义 Hash 函数
     */
    private class MyHash {
        private int cap;
        private int seed;

        MyHash(int cap, int seed) {
            this.cap = cap;
            this.seed = seed;
        }

        /**
         * 计算 Hash 值
         */
        int hash(Object obj) {
            return (obj == null) ? 0 : Math.abs(seed * (cap - 1) & (obj.hashCode() ^ (obj.hashCode() >>> 16)));
        }
    }

    public static void main(String[] args) {
        String s1 = "Hello";
        MyBloomFilter myBloomFilter = new MyBloomFilter();
        System.out.println("s1是否存在：" + myBloomFilter.contains(s1));
        myBloomFilter.add(s1);
        System.out.println("s1是否存在：" + myBloomFilter.contains(s1));
    }
}

4.2 缓存击穿

缓存击穿问题也叫热点Key问题，如果缓存中的某个热点数据过期了，此时大量的请求访问了该热点数据，就无法从缓存中读取，直接访问数据库，数据库很容易就被高并发的请求冲垮，这就是缓存击穿的问题。

4.2.1 添加互斥锁(分布式锁)

保证缓存与数据库的一致性，但是如果缓存重建时间过长，性能会有极大影响，甚至有死锁的风险，牺牲了可用性

当业务线程在处理用户请求时，如果发现访问的数据不在 Redis 里，就加个互斥锁，保证同一时间内只有一个请求来构建缓存（从数据库读取数据，再将数据更新到 Redis 里），当缓存构建完成后，再释放锁。未能获取互斥锁的请求，要么等待锁释放后重新读取缓存，要么就返回空值或者默认值。

public class GoodsService {

    ...
  
    /**
     * 详情操作
     * 分布式锁解决缓存击穿
     * @param id
     * @return
     */
    private Goods query(Long id) {
        // 1.从缓存中获取
        Goods goods = queryByCache(id);
        if (goods != null) {
            return goods;
        }
        //goodsJson 此时一定是 null
        // 2.实现缓存重建
        String key = RedisConstants.CACHE_GOODS_KEY + id;
        // 2.1 获取互斥锁
        RLock rLock = redissonClient.getLock(key);
        try {
            // 2.2判断是否获取成功
            // 分布式锁能保证多个线程并发读取时,只有一个线程能更新缓存
            if (!rLock.tryLock(200, TimeUnit.MILLISECONDS)) {
                // 2.3失败,则休眠并重试
                Thread.sleep(50);
                // 注意：获取锁的同时应该再次检测 redis 缓存是否存在，做 Double Check,如果存在则无需重建缓存
                return queryByCache(id);
            }
            // 2.4成功，根据id查询数据库
            goods = getById(id);
            // 3.不存在，返回错误
            if (goods == null) {
                // 将空值写入到redis
                // RedisConstants.CACHE_EMPTY = ""
                cacheService.set(RedisConstants.CACHE_GOODS_KEY + id, RedisConstants.CACHE_EMPTY,
                        RedisConstants.CACHE_NULL_TTL, TimeUnit.MINUTES);
                return null;
            }
            // 4.存在就加入到Redis,并返回
            cacheService.set(RedisConstants.CACHE_GOODS_KEY + id, JSONUtils.toJsonStr(goods),
                    RedisConstants.CACHE_GOODS_TTL + getRandomNum(), TimeUnit.MINUTES);
        } catch (Exception e) {
            // log error
        } finally {
            // 5.释放互斥锁
            rLock.unlock();
        }
        return goods;
    }

    private Goods queryByCache(Long id) {
        // 1.从查询Redis中是否有数据
        String goodsJson = cacheService.get(RedisConstants.CACHE_GOODS_KEY + id);
        // 2.判断是否存在
        if (StringUtils.isNotBlank(goodsJson)) {
            // 存在则直接返回
            Goods goods = JSONUtils.toBean(goodsJson, Goods.class);
            return goods;
        }
        // 3.判断命中的是否是空值,不是null则一定是空串""
        // RedisConstants.CACHE_EMPTY = ""
        if (RedisConstants.CACHE_EMPTY.equals(goodsJson)) {
            return null;
        }
        return null;
    }

    /**
     * 更新操作需加分布式锁
     * @param goods
     */
    public void update(Goods goods) {
        // 1.get lock
        String key = RedisConstants.CACHE_GOODS_KEY + goods.getId();
        RLock lock = redissonClient.getLock(key);
        try {
            boolean lockFlag = lock.tryLock(2, TimeUnit.SECONDS);
            if (!lockFlag) {
                // 2.update fail
                throw new RuntimeException("update fail");
            }
            // 3.update db
            updateById(goods.getId(), goods);
            cacheService.delete(key);
        } catch (Exception e) {
            // log error
        } finally {
            lock.unlock();
        }
    }

    /**
     * 添加操作不用加分布式锁
     *
     * @param goods
     */
    public void add(Goods goods) {
        save(goods);
    }

    ...
}

4.2.2 数据预热

不给热点数据设置过期时间或者针对热点数据提前预热，将其存入缓存中并设置合理的过期时间比如秒杀场景下的数据在秒杀结束之前不过期。

4.3 缓存雪崩

缓存雪崩是指在同一时段大量的缓存key同时失效或者Redis服务宕机，导致大量请求到达数据库，带来巨大压力。

4.3.1 Redis 服务不可用的情况

采用 Redis 集群，避免单机出现问题整个缓存服务都没办法使用。
给缓存业务添加降级限流策略，当redis故障时，服务及时降级，如快速失败、拒绝服务等。
给业务添加多级缓存，比如说可以在反向代理服务器nginx、JVM内部建立本地缓存来缓解

4.3.2 大量数据同时过期的情况

我们可以在对缓存数据设置过期时间时，给这些数据的过期时间加上一个随机数

互斥锁方案，保证同一时间只有一个业务线程更新缓存，未能获取互斥锁的请求，要么等待锁释放后重新读取缓存，要么就返回空值或者默认值

目录CONTENT

缓存更新和同步策略、缓存穿透、缓存击穿、缓存雪崩