Redis

简单来说 Redis 就是一个使用 C 语言开发的数据库，不过与传统数据库不同的是 Redis 的数据是存在内存中的，也就是它是内存数据库，所以读写速度非常快，因此 Redis 被广泛应用于缓存方向。

Redis的特点

Redis 支持更丰富的数据类型（支持更复杂的应用场景）。Redis 不仅仅支持简单的 k/v 类型的数据，同时还提供 list，set，zset，hash 等数据结构的存储。Memcached 只支持最简单的 k/v 数据类型。
Redis 支持数据的持久化，可以将内存中的数据保持在磁盘中，重启的时候可以再次加载进行使用,而 Memecache 把数据全部存在内存之中。
Redis 有灾难恢复机制。因为可以把缓存中的数据持久化到磁盘上。
Redis 在服务器内存使用完之后，可以将不用的数据放到磁盘上。但是，Memcached 在服务器内存使用完之后，就会直接报异常。
Redis 支持发布订阅模型、Lua 脚本、事务等功能。

缓存数据的处理流程是怎么样的

1. 如果用户请求的数据在缓存中就直接返回。

2. 缓存中不存在的话就看数据库中是否存在。

3. 数据库中存在的话就更新缓存中的数据。

4. 数据库中不存在的话就返回空数据。

为什么要使用Redis/为什么要使用缓存

简单来说主要是为了提升用户体验和能够处理更多的用户。

可以从两个方面解释

高性能

假如用户第一次访问数据库中的某些数据的话，这个过程是比较慢，毕竟是从硬盘中读取的。但是，如果说，用户访问的数据属于高频数据并且不会经常改变的话，那么我们就可以很放心地将该用户访问的数据存在缓存中。

这样有什么好处呢？那就是保证用户下一次再访问这些数据的时候就可以直接从缓存中获取了。操作缓存就是直接操作内存，所以速度相当快。

不过，要保持数据库和缓存中的数据的一致性。如果数据库中的对应数据改变的之后，同步改变缓存中相应的数据即可！

高并发

一般像 MySQL 这类的数据库的 QPS 大概都在 1w 左右（4 核 8g），但是使用 Redis 缓存之后很容易达到 10w+，甚至最高能达到 30w+（就单机 redis 的情况，redis 集群的话会更高）。

QPS（Query Per Second）：服务器每秒可以执行的查询次数；

所以，直接操作缓存能够承受的数据库请求数量是远远大于直接访问数据库的，所以我们可以考虑把数据库中的部分数据转移到缓存中去，这样用户的一部分请求会直接到缓存这里而不用经过数据库。进而，我们也就提高的系统整体的并发。

Redis单线程模型

Redis 基于 Reactor 模式来设计开发了自己的一套高效的事件处理模型，这套事件处理模型对应的是 Redis 中的文件事件处理器（file event handler）。由于文件事件处理器（file event handler）是单线程方式运行的，所以我们一般都说 Redis 是单线程模型。

单线程，怎么监听大量的客户端连接呢？

Redis 基于 Reactor 模式开发了自己的网络事件处理器：这个处理器被称为文件事件处理器（file event handler）。文件事件处理器使用 I/O 多路复用（multiplexing）程序来同时监听多个套接字，并根据套接字目前执行的任务来为套接字关联不同的事件处理器。

当被监听的套接字准备好执行连接应答（accept）、读取（read）、写入（write）、关闭（close）等操作时，与操作相对应的文件事件就会产生，这时文件事件处理器就会调用套接字之前关联好的事件处理器来处理这些事件。

虽然文件事件处理器以单线程方式运行，但通过使用 I/O 多路复用程序来监听多个套接字，文件事件处理器既实现了高性能的网络通信模型，又可以很好地与 Redis 服务器中其他同样以单线程方式运行的模块进行对接，这保持了 Redis 内部单线程设计的简单性。

Redis给缓存数据设置过期时间有啥用

因为内存是有限的，如果缓存中的所有数据都是一直保存的话，分分钟直接Out of memory。

很多时候，我们的业务场景就是需要某个数据只在某一时间段内存在，比如我们的短信验证码可能只在1分钟内有效，用户登录的 token 可能只在 1 天内有效。

如果使用传统的数据库来处理的话，一般都是自己判断过期，这样更麻烦并且性能要差很多。

Redis是如何判断一个数据是否过期？

Redis 通过一个叫做过期字典（可以看作是hash表）来保存数据过期的时间。过期字典的键指向Redis数据库中的某个key(键)，过期字典的值是一个long long类型的整数，这个整数保存了key所指向的数据库键的过期时间（毫秒精度的UNIX时间戳）。

过期数据的删除策略

1. 惰性删除：只会在取出key的时候才对数据进行过期检查。这样对CPU最友好，但是可能会造成太多过期 key 没有被删除。

2. 定期删除：每隔一段时间抽取一批 key 执行删除过期key操作。并且，Redis 底层会通过限制删除操作执行的时长和频率来减少删除操作对CPU时间的影响。

Redis内存的淘汰机制

volatile-lru（least recently used）：从已设置过期时间的数据集（server.db[i].expires）中挑选最近最少使用的数据淘汰
volatile-ttl：从已设置过期时间的数据集（server.db[i].expires）中挑选将要过期的数据淘汰
volatile-random：从已设置过期时间的数据集（server.db[i].expires）中任意选择数据淘汰
allkeys-lru（least recently used）：当内存不足以容纳新写入数据时，在键空间中，移除最近最少使用的 key（这个是最常用的）
allkeys-random：从数据集（server.db[i].dict）中任意选择数据淘汰
no-eviction：禁止驱逐数据，也就是说当内存不足以容纳新写入数据时，新写入操作会报错。这个应该没人使用吧！
volatile-lfu（least frequently used）：从已设置过期时间的数据集(server.db[i].expires)中挑选最不经常使用的数据淘汰
allkeys-lfu（least frequently used）：当内存不足以容纳新写入数据时，在键空间中，移除最不经常使用的 key

Redis持久化机制（怎么保证Redis挂了之后再重启数据不丢失）

Redis 不同于 Memcached 的很重要一点就是，Redis 支持持久化，而且支持两种不同的持久化操作。Redis 的一种持久化方式叫快照（snapshotting，RDB），另一种方式是只追加文件（append-only file, AOF）。这两种方法各有千秋，下面我会详细这两种持久化方法是什么，怎么用，如何选择适合自己的持久化方法。

快照（snapshotting）持久化（RDB）

Redis 可以通过创建快照来获得存储在内存里面的数据在某个时间点上的副本。Redis 创建快照之后，可以对快照进行备份，可以将快照复制到其他服务器从而创建具有相同数据的服务器副本（Redis 主从结构，主要用来提高 Redis 性能），还可以将快照留在原地以便重启服务器的时候使用。

AOF（append-only file）持久化

与快照持久化相比，AOF 持久化的实时性更好，因此已成为主流的持久化方案。默认情况下 Redis 没有开启 AOF（append only file）方式的持久化，可以通过 appendonly 参数开启

开启 AOF 持久化后每执行一条会更改 Redis 中的数据的命令，Redis 就会将该命令写入硬盘中的 AOF 文件。AOF 文件的保存位置和 RDB 文件的位置相同，都是通过 dir 参数设置的，默认的文件名是 appendonly.aof。

appendfsync always #每次有数据修改发生时都会写入AOF文件,这样会严重降低Redis的速度

appendfsync everysec #每秒钟同步一次，显示地将多个写命令同步到硬盘

appendfsync no #让操作系统决定何时进行同步

Redis事物

redis是作为缓存主要使用的，Redis事物不像是数据库事物，它没有原子性，也不支持回滚。

缓存穿透

缓存穿透说简单点就是大量请求的 key 根本不存在于缓存中，导致请求直接到了数据库上，根本没有经过缓存这一层。举个例子：某个黑客故意制造我们缓存中不存在的 key 发起大量请求，导致大量请求落到数据库。

解决方法

1. 缓存无效Key

如果缓存和数据库都查不到某个key的数据就在Redis中写入且设置一个过期时间。这种方式可以解决key变化不频繁的情况，如果有大量不同的key请求来，就会产生大量无效的key，治标不治本。

2. 布隆过滤器

布隆过滤器是一个非常神奇的数据结构，通过它我们可以非常方便地判断一个给定数据是否存在于海量数据中。我们需要的就是判断 key 是否合法，有没有感觉布隆过滤器就是我们想要找的那个“人”。

具体是这样做的：把所有可能存在的请求的值都存放在布隆过滤器中，当用户请求过来，先判断用户发来的请求的值是否存在于布隆过滤器中。不存在的话，直接返回请求参数错误信息给客户端，存在的话才会走下面的流程。

是，需要注意的是布隆过滤器可能会存在误判的情况。总结来说就是：布隆过滤器说某个元素存在，小概率会误判。布隆过滤器说某个元素不在，那么这个元素一定不在。

为什么会出现误判的情况呢? 我们还要从布隆过滤器的原理来说！

我们先来看一下，当一个元素加入布隆过滤器中的时候，会进行哪些操作：

1. 使用布隆过滤器中的哈希函数对元素值进行计算，得到哈希值（有几个哈希函数得到几个哈希值）。

2. 根据得到的哈希值，在位数组中把对应下标的值置为 1。

我们再来看一下，当我们需要判断一个元素是否存在于布隆过滤器的时候，会进行哪些操作：

1. 对给定元素再次进行相同的哈希计算；

2. 得到值之后判断位数组中的每个元素是否都为 1，如果值都为 1，那么说明这个值在布隆过滤器中，如果存在一个值不为 1，说明该元素不在布隆过滤器中。

然后，一定会出现这样一种情况：不同的字符串可能哈希出来的位置相同。（可以适当增加位数组大小或者调整我们的哈希函数来降低概率）

代码实现加测试：

/**
 * 布隆过滤器演示
 *
 * @author 劝书
 * @date 2023/03/28
 */
public class BloomFilterDemo {
    /**
     * 位数组的大小
     */
    private static int SIZE;

    /**
     * 通过这个数组创建不同的哈希函数
     */
    private static int[] SEEDS;

    /**
     * 位数组，只能存0，1
     */
    private BitSet bits;

    /**
     * 存放包含hash函数的类的数组
     */
    private SimpleHash[] fnc;

    /**
     * 误判率
     */
    private MisjudgmentRate rate;

    /**
     * 自动清空
     */
    private  Double autoClearRate;

    /**
     * 使用数量
     */
    private final AtomicInteger useCount = new AtomicInteger(0);

    /**
     * hash函数
     *
     * @author 劝书
     * @date 2023/03/28
     */
    public static class SimpleHash{

        private final int cap;
        private final int seed;

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

        /**
         * 获得哈希值
         *
         * @param value 价值
         * @return int
         */
        public int hash(Object value){
            int h;
            //取低16位
            return (value == null)? 0 : Math.abs(seed * (cap - 1) & ((h = value.hashCode()) ^ (h >> 16)));
        }
    }

    /**
     * 误判率
     *
     * @author 劝书
     * @date 2023/03/28
     */
    public enum MisjudgmentRate {
        /**
         * 每个字符串分配4个位
         */
        VERY_SMALL(new int[] { 1, 2, 3, 4 }),
        /**
         * 每个字符串分配8个位
         */
        SMALL(new int[] { 2, 3, 5, 7, 11, 13, 17, 19 }),
        /**
         * 每个字符串分配16个位
         */
        MIDDLE(new int[] { 2, 3, 5, 7, 11, 13, 17, 19, 23, 29, 31, 37, 41, 43, 47, 53 }),
        /**
         * 每个字符串分配32个位
         */
        HIGH(new int[] { 2, 3, 5, 7, 11, 13, 17, 19, 23, 29, 31, 37, 41, 43, 47, 53, 59, 61, 67, 71, 73, 79, 83, 89, 97,
                101, 103, 107, 109, 113, 127, 131 });

        private int[] seeds;

        private MisjudgmentRate(int[] seeds) {
            this.seeds = seeds;
        }

        public int[] getSeeds() {
            return seeds;
        }

        public void setSeeds(int[] seeds) {
            this.seeds = seeds;
        }
    }

    /**
     * 布隆过滤器构造函数，默认使用中等误判率
     *
     * @param dataCount 数据统计
     */
    public BloomFilterDemo(int dataCount){
        this(MisjudgmentRate.VERY_SMALL,dataCount,null);
    }

    /**
     * 布隆过滤器演示
     *
     * @param rate 误判率
     * @param dataCount       预期处理的数据规模，如预期用于处理1百万数据的查重，这里则填写1000000
     * @param autoClearRate   自动清空过滤器内部使用比率
     */
    public BloomFilterDemo(MisjudgmentRate rate,int dataCount,Double autoClearRate){
        long bitSize = (long) rate.seeds.length * dataCount;
        if(bitSize < 0 || bitSize > Integer.MAX_VALUE){
            throw new RuntimeException("位数不合适，请降低误判率或调整数据大小");
        }
        this.rate = rate;
        SEEDS = rate.seeds;
        SIZE = (int) bitSize;
        fnc = new SimpleHash[SEEDS.length];
        for (int i = 0; i < SEEDS.length; i++) {
            fnc[i] = new SimpleHash(SIZE,SEEDS[i]);
        }
        bits = new BitSet(SIZE);
        this.autoClearRate = autoClearRate;
    }


    /**
     * 添加元素到位数组中
     *
     * @param value 值
     */
    public void add(Object value){
        checkNeedClear();

        if(!contains(value)){
            for (SimpleHash f: fnc) {
                bits.set(f.hash(value),true);
            }
            useCount.getAndIncrement();
        }
    }

    /**
     * 判断指定元素是否包含在位数组中
     *
     * @param value 价值
     * @return boolean
     */
    public boolean contains(Object value){
        boolean ret = true;
        for(SimpleHash f: fnc){
            ret = ret && bits.get(f.hash(value));
        }
        return ret;
    }

    /**
     * 检查是否需要清空
     */
    private void checkNeedClear(){
        if(autoClearRate != null){
            if(getUseRate() >= autoClearRate){
                synchronized (this){
                    if (getUseRate() >= autoClearRate){
                        bits.clear();
                        useCount.set(0);
                    }
                }
            }
        }
    }

    /**
     * 得到利用率
     *
     * @return double
     */
    public double getUseRate(){
        return (double) useCount.intValue() / (double) SIZE;
    }

    /**
     * 测试
     */
    public static void main(String[] args) {
        String v1 = "123321";
        String v2 = "321123";

        BloomFilterDemo filter = new BloomFilterDemo(2<<24);
        System.out.println(filter.contains(v1));
        System.out.println(filter.contains(v2));
        filter.add(v1);
        filter.add(v2);
        System.out.println(filter.contains(v1));
        System.out.println(filter.contains(v2));
    }
}

输出：

false

false

true

true

实现一个布隆过滤器，很nice。

缓存雪崩

实际上，缓存雪崩描述的就是这样一个简单的场景：缓存在同一时间大面积的失效，后面的请求都直接落到了数据库上，造成数据库短时间内承受大量请求。这就好比雪崩一样，摧枯拉朽之势，数据库的压力可想而知，可能直接就被这么多请求弄宕机了。

举个例子：系统的Redis缓存模块出了问题比如宕机导致不可用。造成系统的所有访问，都要走数据库。

还有一种缓存雪崩的场景是：有一些被大量访问数据（热点缓存）在某一时刻大面积失效，导致对应的请求直接落到了数据库上。

举个例子：秒杀开始 12 个小时之前，我们统一存放了一批商品到 Redis 中，设置的缓存过期时间也是 12 个小时，那么秒杀开始的时候，这些秒杀的商品的访问直接就失效了。导致的情况就是，相应的请求直接就落到了数据库上，就像雪崩一样可怕。

解决方法

针对 Redis 服务不可用的情况：

采用 Redis 集群，避免单机出现问题整个缓存服务都没办法使用。
限流，避免同时处理大量的请求。

针对热点缓存失效的情况：

设置不同的失效时间比如随机设置缓存的失效时间。
缓存永不失效。

如何保证缓存和数据库数据的一致性？双删策略

对于数据库做增删改操作，先删除缓存，操作数据库，再删除缓存，重新缓存数据库。

如果更新数据库成功，而删除缓存这一步失败的情况的话，简单说两个解决方案：

缓存失效时间变短（不推荐，治标不治本）：我们让缓存数据的过期时间变短，这样的话缓存就会从数据库中加载数据。另外，这种解决办法对于先操作缓存后操作数据库的场景不适用。
增加cache更新重试机制（常用）：如果 cache 服务当前不可用导致缓存删除失败的话，我们就隔一段时间进行重试，重试次数可以自己定。如果多次重试还是失败的话，我们可以把当前更新失败的 key 存入队列中，等缓存服务可用之后，再将缓存中对应的 key 删除即可。