Redis

Redis用过吗，和本地缓存有啥区别？

Redis 是 分布式缓存 ，本地缓存是 单机缓存 ，那么在分布式系统中，如果将数据放在本地缓存中，其他节点肯定是无法进行访问了

其次就是 本地缓存相对于 Redis 缓存来说会更快 ，因为去 Redis 中查询数据虽然 Redis 基于内存操作比较快，但是应用还需要和 Redis 发起网络 IO ，而使用本地缓存就不需要网络 IO 了，因此本地缓存更快

布隆过滤器了解吗？

说布隆过滤器之前，肯定要知道 BitMap 到底是什么

BitMap 到底用于解决什么问题？

BitMap 常常用于解决一些数据量比较大的问题，比如说对于1千万个整数，整数的范围在 1~100000000，对于一个整数 x ，我们怎么快速知道 x 在不在这1千万个整数中呢？

使用 BitMap 来解决的话，就把存在的整数的位置给设置为 true，比如 arr[k]=true，那么判断整数 x 是否在这1千万个整数中，只需要判断 arr[x] == true 即可。

那既然这样为什么不使用数组来标记呢？因为数组所占空间过大，会导致内存溢出。比如使用 int[] arr = new int[10];，在 java 中，一个 int 占用 4B，4B = 32bit，那如果使用 BitMap，使用 1 个 bit 来标记一个数，BitMap 的空间占用是数组的 1/32

Java 中 BitMap 如何实现了？

Java 中 hutool 工具包中实现了 BitMap，引入Maven依赖为：

<dependency>
    <groupId>cn.hutool</groupId>
    <artifactId>hutool-all</artifactId>
    <version>5.8.0.M2</version>
</dependency>

Java 中有两种 bitmap，分别为 IntMap、LongMap，这里以 IntMap 为例：

初始时，ints 数组即为 bitmap 数组，如果我们要向数组中添加 5，过程如下：

• 第一步：计算在数组中的下标，数组中一个 int 数有 32 位，可以存储 32 个标记，5 < 32，所以在第一个 int 数中存储，计算公式：5 / 32 = 0
• 第二部：计算 5 在当前这个 int 数中存储的位置，也就是在当前这个 32 个位置中，处于第几个位置，所以和 31 进行与操作，31 也就是全1，通过和全1进行与操作就可以计算出位置，计算公式：5 % 32 = 5

上边计算出来了下标为 0，在 0 个 int 数的第 5 个位置存储

图片解释如下：

public class IntMap implements BitMap, Serializable {
    private static final long serialVersionUID = 1L;
    private final int[] ints;

    public IntMap() {
        this.ints = new int[93750000];
    }

    public IntMap(int size) {
        this.ints = new int[size];
    }

    public void add(long i) {
        int r = (int)(i / 32L);
        int c = (int)(i & 31L);
        this.ints[r] |= 1 << c;
    }

    public boolean contains(long i) {
        int r = (int)(i / 32L);
        int c = (int)(i & 31L);
        return (this.ints[r] >>> c & 1) == 1;
    }

    public void remove(long i) {
        int r = (int)(i / 32L);
        int c = (int)(i & 31L);
        int[] var10000 = this.ints;
        var10000[r] &= ~(1 << c);
    }
}

BitMap总结：

• BitMap适合存储数据密集的数据，如果对于稀疏数据会造成空间浪费。
• 相比于其他数组更加节省空间

BitMap的一些适用场景：

• 统计指定用户一年中的登陆天数，哪几天登陆了，哪几天没有登陆？
• 判断用户的在线状态
• 统计活跃用户，使用时间作为缓存的key，用户id值为value中的偏移量，这一时间段如果活跃就设置为1

接下来看一下为什么有了 BitMap 之后，还需要使用布隆过滤器呢？

对于仅仅为整数的判断可以使用 BitMap 来进行实现，那么我们来考虑下边这个场景：

如果网站需要设置一个黑名单，url数量会上亿，我们怎么将不是整数的 url 给存储进 BitMap 中呢？存储在 BitMap 中的下标该如何计算呢？

对于这些复杂情况，就可以使用 Redis 的一种数据结构布隆过滤器，首先布隆过滤器的特点就是判断一个 url 的哈希值的位置，如果是1，则可能在集合中，但是如果不是1，则一定不在集合中。

这是为什么呢？

因为布隆过滤器使用了一组哈希函数，如果仅仅使用一个哈希函数，当 url 数量变多时，计算出来的哈希值冲突概率极高，假设使用的一组哈希函数为h1,h2,...hk，那么布隆过滤器会对一个 url 计算 k 次哈希值，得到 k 个哈希值，将 BitMap 中这 k 个位置都设置为1，那么在判断一个 url 是否存在时，判断这 k 个位置的值是否全部为1，如果有一个位置不为1，则表示该 url 并不在集合中。

布隆过滤器总结：

• 时间、空间效率高
• 误判率低，可以通过调整哈希函数数量和数组大小来控制误判率
• 支持并发
• 无法确定元素一定集合中！
• 常用于作为一层程序拦截

项目中布隆过滤器怎么用的？为啥要用，不用行不行？

一般是项目中用到了布隆过滤器，面试官提问的概率会大一些，如果项目中没有使用的话，可能不会问到，不过也可以将布隆过滤器加入到项目中，作为一个小亮点

使用 布隆过滤器 一般就是用于快速判断某个元素是否在集合中出现了，可以用于解决 缓存穿透 的问题，布隆过滤器提供 一组哈希函数 h1, h2, ..., hk ，对需要存储的数据使用哈希函数计算得到 k 个哈希值，将 BitMap 中这 k 个位置都设置为 1，如果这 k 个位置都是1，则 可能 在集合中，但是如果都不是1，则 一定不在 集合中

因此布隆过滤器会出现 误判 ，可能将不在集合中的元素判断为在集合中，可以通过 增加数组长度 来降低误判率

缓存穿透： 请求的数据在数据库中不存在，因此数据也不会在缓存中，每次请求都不会命中缓存，而是打到数据库上，也就是直接穿过缓存打到数据库中，导致数据库压力很大甚至崩溃，这就是缓存穿透

那么缓存穿透的话，可以使用 Redis 的 布隆过滤器 来解决：下载 RedisBloom 插件，该插件是 Redis 的布隆过滤器模块，下载之后在 Redis 的 conf 文件中配置之后即可使用

具体解决缓存穿透的场景 的话，这里举一个例子： 用户注册场景 ，如果系统用户量很大，在用户注册的时候，需要判断用户的用户名是否重复，初始将用户名的信息都初始化在布隆过滤器中，那么在用户注册时，先去布隆过滤器中快速进行判断用户名是否已经被使用，如果经过 k 次哈希计算发现这 k 次哈希值的位置上都为 1，说明 该用户名可能已经被使用了 ，用户注册用户名重复的话，大不了就换一个用户名就好了，这种情况是可以容忍的，之后用户注册成功之后，再将注册成功的用户名也放入的布隆过滤器中，这样在 用户注册时可以通过布隆过滤器快速判断用户名是否重复

上边说了布隆过滤器可能存在 误判 的情况，误判是可以容忍的，但是布隆过滤器解决缓存穿透还存在另外一个缺点：无法删除元素

无法删除元素 会导致如果用户注销帐号了，那么该用户名是无法从布隆过滤器中删除的，因此会导致其他用户也无法注册这个用户名，可以考虑再添加一层 Redis 缓存 来存储已经注销的用户名，同时如果注销的用户名较多的话，可能存在 大 key 问题 ，可以考虑分片存储来解决

这里总结一下如何通过 布隆过滤器解决缓存穿透：

首先将用户名都初始化在布隆过滤器中，用户注册的时候通过 布隆过滤器 快速判断该用户名是否已经被使用了，系统可以容忍一定的误判率，对于布隆过滤器无法删除元素这个缺点，添加一层 Redis 缓存，将已经注销的用户名放在这个 Redis 中的 set 里，这样就可以解决布隆过滤器无法删除元素的缺点了，不过如果注销用户名多了，可能会存在大 key 的问题，因此要考虑 分片存储 解决大 key 问题，也可以从业务角度上，限制每个用户注销的次数

最后再说一下布隆过滤器中容量的计算：

先说一下各个参数的含义：

• m： 布隆过滤器中二进制 bit 数组的长度
• n ：需要对多少个元素进行存储，比如说我们要存储 1000 万个用户名，那么 n = 1000 万
• p： 期望的误判率，可以设置 p = 0.001（%0.1） 或者 p = 0.0001（%0.01）

将 n、p 带入上述公式即可计算出来理想情况下布隆过滤器的二进制数组的长度，也可以根据此公式算出来存储这么多元素大概需要占用多少内存空间，比如需要存储 10 亿个用户名，期望误判率为 0.001，也就是将 n = 10亿、p = 0.001 带入，得到 m 约为 1.67GB ，因此这个布隆过滤器大约占用 1.67GB 的空间（可以搜索在线布隆过滤器容量计算）

了解布谷鸟过滤器？

布谷鸟过滤器（Cuckoo Filter）也是是一种数据结构，相比于布隆过滤器来说，添加了可以删除元素的功能

下边说一个简单的例子，假设我们有一个包含四个槽位（桶）的布谷鸟过滤器，以及两个哈希函数 h1 和 h2，如果此时去向布谷鸟过滤器中添加一个元素 "A"，

1. 初始化：首先，我们初始化四个槽位，所有槽位都是空的。
2. 添加元素：
   • 假设我们要添加元素 "A"。我们使用哈希函数 h1 和 h2 对 "A" 进行哈希，得到槽位位置 h1("A") 和 h2("A")。
   • 如果这两个槽位都是空的，我们就将 "A" 放入这些槽位中。
   • 如果 h1("A") 的槽位已经被占用，我们将原来的元素 "B" 踢出，将元素 "B" 通过 h2 哈希到另一个槽位。如果这个新槽位也是空的，我们就将 "B" 放入；如果不是，我们继续踢出原有的元素并重新哈希，直到找到一个空槽位。
3. 查询元素：
   • 要检查元素 "A" 是否存在，我们再次使用 h1 和 h2 对 "A" 进行哈希，检查对应的槽位是否有 "A"。
   • 如果两个槽位中至少有一个有 "A"，我们认为 "A" 可能存在（因为存在误判的可能性）。
4. 删除元素：
   • 要删除元素 "A"，我们同样使用 h1 和 h2 对 "A" 进行哈希，找到对应的槽位。
   • 如果槽位中有 "A"，我们将其移除，并将槽位标记为空。这个过程不需要踢出其他元素，因为删除操作不会影响其他元素的位置。
5. 处理冲突：
   • 在添加或删除过程中，如果连续尝试了一定次数（例如，迭代次数）后仍然找不到空槽位，我们认为发生了冲突，可能需要扩展过滤器的大小或者报告添加失败。

布谷鸟过滤器在需要频繁进行元素添加和删除的场景中特别有用，例如在缓存系统中，它可以帮助减少因缓存穿透和缓存击穿导致的数据库

了解 Redis 的 IO 多路复用吗？

我们都知道 Redis 性能是很高的，它的高性能是基于 两个方面 ：

• 基于 内存 操作
• IO 多路复用

基于内存操作比较容易理解，将所有的操作都放在 内存 中进行处理即可，所以对于 Redis 来说，内存可能会成为 性能的瓶颈

Redis 采用的 IO 多路复用模型

第二个方面就是 Redis 中所使用的 IO 模型：IO 多路复用

IO 多路复用 是一种 IO 模型，它的思想是：使用单个线程来处理多个客户端连接

多路 即同时监听多个客户端连接通道上的事件，复用 即使用单个线程处理多个客户端连接，复用同一个线程去处理请求，而不是重复创建线程

Redis 基于 IO 多路复用 的思想，实现了通过 单个线程 处理大量客户端请求的功能，Redis 中的 IO 多路复用 如下图：

当应用想要与 Redis 进行通信的话，就会和 Redis 建立 TCP 网络连接，客户端与服务端之间通信就是靠 Socket 来完成的，Socket 就是对底层操作系统进行网络通信封装的一组 抽象接口

那么 Redis 的 IO 多路复用指的就是在 Redis 的服务端，他会通过 一个线程 来处理 多个 Socket 中的事件，包括连接建立、读、写事件等，当有 Socket 发生了事件，就会将该 Socket 加入到 任务队列 中等待事件分发器来处理即可

这里的 IO 多路复用 的核心就是通过一个线程处理多个 Socket 上的事件，常见的 IO 多路复用底层实现 有：select、poll、epoll（这里就先略过底层的实现了，主要看整体的 IO 多路复用模型）

上边只解释了 IO 多路复用是什么，Redis 使用这个 IO 模型肯定是因为它快，因此要将 IO 多路复用和其他的一些IO 模型进行对比，才能知道它到底 快在哪里

常见的 IO 模型有：BIO、NIO、AIO 这三种 IO 模型：（这里主要说一下 BIO 和 NIO，AIO 还没有广泛使用，常用的 Netty 就是基于 NIO 的）

• BIO

先说一下 BIO ：BIO 是同步阻塞式的：当客户端有连接发送到服务端时，服务端会为每一个客户端都创建一个线程来进行处理

那么它的 问题 有两个：

1、在并发量很大的情况下，服务端需要创建大量的线程来处理客户端请求，占用大量系统资源

2、一个线程监听一个 Socket，如果该 Socket 上没有事件的话，线程会一直阻塞等待，造成资源浪费

• NIO

那么 BIO 的缺点显而易见，需要 创建大量线程 ，并且线程会阻塞等待

接下来说一下 NIO，它是基于 IO 多路复用 实现非阻塞的，主要有 3 个核心概念：Channel、Buffer、Selector

Selector： 就是 IO 多路复用中去监听多个 Socket 的线程

Channel： 就是和客户端之间建立连接的通道

Buffer： 在 NIO 中数据传输时依靠 Buffer 进行传输的，也就是数据会从 Channel 读取到 Buffer 缓冲区，再拿到 Buffer 中的数据进行处理

那么 NIO 基于 IO 多路复用所带来的好处就是通过单线程就可以监听处理大量的客户端连接、请求事件，不需要创建大量线程，也不会阻塞很长时间，导致资源浪费

总结

因此 Redis 对于连接、读写事件的处理就是基于 IO 多路复用 来做的，这样通过单个线程来处理多个客户端连接中发生的事件，既节省系统资源，而且 基于内存 操作，处理起来速度是比较快的

这里最后说一下 IO 多路复用、select、poll、epoll、Reactor 模式之间的关系： IO 多路复用的底层实现有 select、poll、epoll，而 Reactor 模式是对 IO 多路复用的封装，更方便使用，Reactor 模式又分为了单 Reactor 单线程模型，单 Reactor 多线程模型、主从 Reactor 模型

扩展：Redis6.0 引入多线程

这里提一下 Redis6.0 引入的多线程（默认关闭）：目的是为了提高 网络IO 的处理性能

• Redis 的多线程只是用来处理网络请求

• 对于读写命令的执行，还是使用单线程进行处理，因此并不存在线程安全的问题

为什么要引入多线程呢，因为网络 IO 的处理是比较慢的，引入的多线程就是下图中的 IO 线程，主要对 Socket 中的请求事件解析，解析之后还是交给单线程进行命令的执行

Redisson 分布式锁？在项目中哪里使用？多久会进行释放？如何加强一个分布式锁？

答：

什么时候需要使用分布式锁呢？

在分布式的场景下，使用 Java 的单机锁并不可以保证多个应用的同时操作共享资源时的安全性，需要通过分布式锁保证多个进程同步访问共享资源

举一个使用分布式锁的场景吧？

在积分场景下需要使用分布式锁，因为积分属于是共享资源，需要保证多个应用对积分的同步访问才行，那么如下图，则为不添加分布式锁时，如何造成了数据不安全：

首先入门级别的分布式锁是通过 Setnx 进行实现，使用 Setnx 实现有四个注意点

1. 需要设置锁的超时时间（如果不设置，在释放锁时，如果机器宕机，会导致锁无法释放）

2. 需要设置一个唯一 ID，表示这个锁是哪个用户添加的，必须由添加锁的用户释放

   （如果不设置，线程1在执行任务时，可能锁的超时时间已经达到，被自动释放，此时线程2加锁，开始执行业务，但正好线程1执行完毕，释放锁，由于没有唯一ID表示，线程1将线程2加的锁给释放掉了）

3. 需要锁续命

   有可能锁的过期时间设置的太短，导致业务没有执行完毕，锁就被自动释放，因此要使用锁续命来解决（大概逻辑是使用子线程执行定时任务，定时任务间隔时间要小于 key 的过期时间，子线程隔一段时间判断主线程是否在执行，如果在执行，就重新设置一下过期时间）

4. 可重入问题：Setnx 实现的分布式锁不可重入，这样获取锁的线程在重复进入相同锁的代码块中会造成死锁

而在 Redission 中已经帮我们实现好了分布式锁，下来看一下 Redission 中的分布式锁：

Redission 中获取锁逻辑：

在 Redission 中加锁，通过一系列调用会到达下边这个方法

他的可重入锁的原理也就是使用 Hash 结构来存储锁，key 表示锁是否存在，如果已经存在，表示需要重复访问同一把锁，会将 value + 1，即每次重入一次 value 就加 1，退出一次 value 就减 1

比如执行以下加锁命令

RLock lock = Redisson.getLock("lock");
lock.lock();

加锁最终会走到下边这个方法，在执行 lua 脚本时有三个参数分别为：

• KEYS[1]： 锁名称，也就是上边的 "lock"
• ARGV[1]： 锁失效时间，默认 30 s
• ARGV[2]： 格式为id + “:” + threadId:锁的小key，值为 c023afb1-afaa-402a-b23e-a21a82abec9d:1

这里讲解下边这个 lua 脚本的执行流程：

1. 先去判断 KEYS[1] 这个哈希结构是否存在

2. 如果不存在，通过 hSet 去创建一个哈希结构，并放入一个 k-v 对

   这个哈希表名为锁的名称，也就是 "lock"，key 为 ARGV[2]，也就是 c023afb1-afaa-402a-b23e-a21a82abec9d:1， value 为 1

3. 通过 pexpire 设置 key 的过期时间，pexpire 的过期时间单位是毫秒，expire 单位是秒

4. 如果这个哈希结构存在，去判断这个 key 是否存在

5. 如果这个 key 存在，表示之前已经被当前线程加过锁了，再去重入加锁即可，也就是通过 hincrby 给这个 key 的值加 1 即可，并且设置过期时间

<T> RFuture<T> tryLockInnerAsync(long leaSetime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
    internalLockLeaSetime = unit.toMillis(leaSetime);

    return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
              "if (Redis.call('exists', KEYS[1]) == 0) then " +
                  "Redis.call('hSet', KEYS[1], ARGV[2], 1); " +
                  "Redis.call('pexpire', KEYS[1], ARGV[1]); " +
                  "return nil; " +
              "end; " +
              "if (Redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                  "Redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
                  "Redis.call('pexpire', KEYS[1], ARGV[1]); " +
                  "return nil; " +
              "end; " +
              "return Redis.call('pttl', KEYS[1]);",
                Collections.<Object>singletonList(getName()), internalLockLeaSetime, getLockName(threadId));
}

Redission 中锁续命原理：

Redission 底层有个看门狗机制，加锁成功后会有一个定时任务，默认锁的失效时间是 30s，该定时任务每隔锁失效时间的 1/3 就会去续约锁时间，也就是每隔 10s 进行锁续命

如何加强一个分布式锁？

也就是如何提升一个分布式锁的性能，分布式锁本质上是将并行操作改为串行，那么我们可以通过使用分段锁来提升性能，比如说有 1000 个库存的话，读入到缓存中将分为 10 份进行存储，即 product_stock_1 = 100, product_stock_2 = 100, ...，每一份库存都各自有一把锁，那么多个线程来竞争这 10 把锁，比原来竞争 1 把锁的性能提高 10 倍

Zset 几个命令的时间复杂度？

答：

• zadd：O(logn)，添加一个元素的时间复杂度是 O(logn)（因为插入元素的话，时间开销都在查找插入位置上，在 Zset 中，查找时间复杂度是 O(logn)，因此插入复杂度同是）
• zrange：O(logn + m) ，n 是集合中元素数量，m 是指定范围内的用户数量

Redis 里面的命令，比如 Setnx 和 Setex 还有 Zset 中的命令？

答：

Zset 中的常用命令为：

• zadd <key> <score1> <value1> <score2> <value2> ...

  向集合 key 中添加元素

• zrange <key> <start> <stop> [withscores]

  查找下标在 start 和 stop 之间的元素，如果后边带上 withscores 参数，会将分数也查询出来

• zrevrange <key> <start> <stop> [withscores]

  将分数从大到小进行查询，和 zrange 查询顺序相反

• zrangebyscore <key> <min> <max> [withscores]

  返回集合 key 中所有 score 介于 min 和 max 之间的成员，如果后边带上 withscores 参数，会将分数也查询出来

• Setex <key> <seconds> <value>

  设置 key、value 并且设置过期时间

• Setnx <key> <value>

  仅当 key 不存在时，才将 key 的值设置为 value，成功返回1，失败返回0

缓存怎么保证数据的一致性？

答：

这里说一下使用 Redis 缓存 + MySQL 如何保证数据一致性（对于本地缓存 + Redis 缓存也是同理）：

那么保证数据一致性就需要保证数据库和缓存同时进行更新，那么就分为两种情况先更新数据库还是先更新缓存，由于更新缓存成本比较高（因为写入数据库的值有时候不是直接写入缓存而是经过一系列计算之后才写入缓存，因此当数据修改时不更新缓存，直接将缓存删除），那么就分为了 先删除缓存再更新数据库 和 先更新数据库，再删除缓存 两种情况：

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

但是在这种情况下还有可能造成的缓存不一致：线程 A 先删除缓存，再去更新数据库，在线程 A 更新数据库之前，如果线程 B 去读取缓存，发现并不存在，去读取数据库，此时读取的是旧数据，再将旧数据写入缓存，此时缓存存储的就是脏数据了。

使用延时双删可以解决此情况的数据不一致，在延时双删中，会删除两次缓存，第二次删除缓存会先等待一段时间，等待数据库更新完毕之后，可以再次删除缓存，分为以下几步：

1. 删除缓存
2. 更新数据库
3. 睡眠 Thread.sleep() 等待
4. 再删除缓存

即延时双删在线程 A 更新完数据库之后，休眠一段时间，再去删除缓存中可能存在的脏数据。

使用延时双删的话，可以结合 AOP + 延时队列实现（下一个问题会给出实现），因为需要隔一段时间再去删除缓存，可能会导致整个操作耗时过长，因此可以将第二次删除缓存的操作异步化

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

这种情况可能因为线程 A 没有及时删除缓存或者删除缓存失败而导致线程 B 读取到旧数据

因此当缓存删除失败时，可以使用消息队列来重试，流程如下：

1. 先将要删除的缓存值或者是要更新的数据库值暂存到消息队列中
2. 当程序没有成功删除缓存值或者更新数据库值时，从消息队列中读取这些值，再次进行删除或更新
3. 如果成功删除缓存或者更新数据库，要将这些值从消息队列中取出，以免重复操作

上边两种保证数据一致性的方法，操作上比较简单，性能也比较好，但是整个缓存删除的操作和业务代码耦合度比较高并且不能保证严格的一致性，如果需要更严格的一致性保障可以选择通过 Canal + MQ 的组合来保证，但相对应的就是会提升系统的复杂性，可以根据具体需求来进行选择

Canal + MQ 方式保证数据一致性：：通过 Canal + RocketMQ 实现缓存数据库的一致性

通过 canal + RocketMQ 来实现数据库与缓存的最终一致性，对于数据直接更新 DB 的情况，通过 canal 监控 MySQL 的 binlog 日志，并且发送到 RocketMQ 中，MQ 的消费者对数据进行消费并解析 binlog，过滤掉非增删改的 binlog，那么解析 binlog 数据之后，就可以知道对 MySQL 中的哪张表进行 增删改 操作了，那么接下来我们只需要拿到这张表在 Redis 中存储的 key，再从 Redis 中删除旧的缓存即可，那么怎么取到这张表在 Redis 中存储的 key 呢？

可以我们自己来进行配置，比如说监控 sku_info 表的 binlog，那么在 MQ 的消费端解析 binlog 之后，就知道是对 sku_info 表进行了增删改的操作，那么假如 Redis 中存储了 sku 的详情信息，key 为 sku_info:{skuId}，那么我们就可以在一个地方对这个信息进行配置：

// 配置下边这三个信息
tableName = "sku_info"; // 表示对哪个表进行最终一致性
cacheKey = "sku_info:"; // 表示缓存前缀
cacheField = "skuId"; // 缓存前缀后拼接的唯一标识

// data 是解析 binlog 日志后拿到的 key-value 值，data.get("skuId") 就是获取这一条数据的 skuId 属性值
// 如下就是最后拿到的 Redis 的 key
RedisKey = cacheKey + data.get(cacheField)

那么整体的流程图如下：

分布式环境下，本地缓存保证数据一致性：

使用 MQ 来保证

分布式环境下，由于本地缓存只在本应用内部有效，当其中一台应用的本地缓存更新之后，我们需要保证其他应用的本地缓存也同步进行更新，因此需要通过 MQ 来保证最终数据一致性

更新流程为：

当应用 1 收到请求更新数据库，同时应用 1 更新本地缓存，并且发送更新 MQ 广播消息，让其他的应用也更新本地缓存，达到数据一致性

延时双删如何实现？

一般如果要保证数据一致性的话，使用延时双删还是比较多的，并且实现起来流程比较简单

实现思路为：使用 AOP+注解+延时队列 来实现

• 延时双删的流程为：

1、删除缓存

2、更新数据库

3、将删除缓存的任务放在延时队列中

4、再删除缓存

为了使用延时双删保证数据一致性，就要在每一个更新操作中都去 加入延时双删的代码

那么一直去加重复的代码显然是没有意义的，因此可以利用设计模式来优化代码结构，也就是代理模式

我们只需要在更新操作的前后删除缓存，很符合代理模式，对更新操作创建一个代理，将更新操作前后删除缓存的动作给抽取到代理的拦截中即可

在注解中，设置一个属性 name ，用来表示缓存的 key，再设置一个延时时间 delayTime ，表示需要等待多长时间删除第二次缓存

在 AOP 切面中，获取注解中的属性信息，将删除缓存的消息发送到延时队列中，再启动一个异步线程不断去延时队列中消费数据，可以捕获异常，默认重试 3 次，如果 3 次以上还没有删除 key，说明 Redis 可能存在问题，就不一直重试了

（https://blog.csdn.net/qq_45038038/article/details/135945574）

@Slf4j
@Component
public class CacheQueue {
 
    @Resource
    RedissonClient redissonClient;
 
    public static final String QUEUE_KEY = "CACHE-DELAY-QUEUE";
 
    // 延时删除
    public void delayedDeletion(String key, Long time, TimeUnit unit) {
        RBlockingQueue<String> blockingQueue = redissonClient.getBlockingQueue(QUEUE_KEY);
        RDelayedQueue<String> delayedQueue = redissonClient.getDelayedQueue(blockingQueue);
        log.info("延时删除key: {} , 当前时间: {} ", key, LocalDateTime.now().toString());
        delayedQueue.offer(key, time, unit);
    }
 
    // 消息监听
    @PostConstruct
    public void listen() {
        CompletableFuture.runAsync(() -> {
            RBlockingQueue<String> blockingQueue = redissonClient.getBlockingQueue(CacheQueue.QUEUE_KEY);
            while (true) {
                try {
                    consumer(blockingQueue.take());
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
    }
 
    // 消费消息
    public void consumer(String key) {
        log.info("删除key: {} , 当前时间: {} ", key, LocalDateTime.now().toString());
        redissonClient.getBucket("key").delete();
    }
}

了解 Redis 中的大key吗？多大算是大key呢？如何解决？

答：

Redis 的大 key 指的是 key 对应的 value 所占用的内存比较大。

对于 String 类型来说，一般情况下超过 10KB 则认为是大 key；对于Set、Zset、Hash 等类型来说，一般数据超过5000条即认为是大 key

bigkey 除了会消耗更多的内存空间和带宽，还会对性能造成比较大的影响。应该避免系统中出现 bigkey。

查找大key：

1. 使用命令 Redis-cli -p 6379 --bigkeys -i 3 在 Redis 中查找大 key，通过 -i 控制扫描频率，表示扫描过程中每次扫描后休息的时间间隔为 3 秒。
2. 分析 RDB 文件来找出大 key。

解决大key：

1. 将大key切割为多个小key。（不推荐，这样需要修改业务代码）
2. 如果大key不是热点key，手动删除。Redis 4.0 之后提供了 unlink 命令，可以异步删除传入的key

了解 Redis 中的热 key 吗？

答：

热 key 指的是在 Redis 接收的读写请求中，某个 key 就占了一半甚至更多的请求，称为热 key。

处理热 key 会占用大量的 CPU 和带宽，如果在某一时间内大量访问热key，请求数量超过 Redis 的处理能力或者热key正好在缓存中过期，就会导致线上服务崩溃。

因此需要对系统中的热 key 进行优化，确保系统的高可用和稳定性。

查找热key：

1. 使用 Redis-cli --hotkeys 来查找热key，使用该命令时，需要设置 maxmemory-policy 为 LRU 算法，否则会报错。

2. 使用 monitor 命令来实时查看 Redis 实例的操作情况，包括读写删除等操作。

   该命令对 Redis 性能影响较大，谨慎使用。

   使用 monitor 命令并输出重定向至文件，在关闭 monitor 命令后对文件进行分析，即可找到这段时间内的热 key。

解决热key：

1. 统计热 key，并使用 jvm 缓存来存储热 key，降低 Redis 缓存的读压力（jvm 缓存存储可以使用 Caffeine，热 key 检测可以使用京东的 hotkey 框架）
2. 使用 Redis 切片集群：在多个 Redis 实例上都存储一份热 key，分散热 key 的访问请求
3. 读写分离：主节点处理写请求，从节点处理读请求

一般比较常用的方式是 使用 jvm 缓存来存储热 key 数据 ，首先需要热 key 探测系统来探测热 key，探测到之后，将热 key 存入 jvm 缓存中即可，流程如下：

Redis 如何存储热点数据？

热点数据就是某些数据访问频率过高

首先对于热点数据要先可以检测到，检测之后再看具体如何解决

检测的话，可以使用京东零售开源的 hotkey 框架，是一个轻量级通用热 key 探测中间件，可以快速将热数据推送到 JVM 内存，减少大量请求对下游服务、Redis、MySQL 的冲击

那么热点数据的存储的话，可以存储一份到 JVM 内存中，进一步提升数据的查询性能，可以选用 Caffeine （Java 高性能缓存库）来管理 Redis 的热点数据，Caffeine 是基于 Google Guava 改进的，因此 Caffeine 的性能表现、缓存命中率相对来说更好

Redis用的什么版本，6.0多线程体现在哪里？Redis的性能瓶颈在哪里？

答：自己项目中的 Redis 使用的 6.2.6版本。Redis的性能瓶颈在于 内存大小 、 网络 IO 处理速度

Redis6.0 引入了多线程，引入多线程的目的是为了提高网络IO的处理性能

• Redis 的多线程只是用来处理网络请求

• 对于读写命令的执行，还是使用单线程进行处理，因此并不存在线程安全的问题。

6.0 多线程默认关闭，在 Redis.conf 中设置 io-threads-do-reads yes 进行开启，并且设置线程个数io-threads 6

主线程与IO线程的协作流程：

1. 服务端和客户端建立Socket连接，并分配处理线程

   主线程负责接收建立连接请求。当有客户端请求和实例建立Socket连接时，主线程会创建和客户端的连接，并把 Socket 放入全局等待队列中。 紧接着，主线程通过轮询方法把Socket连接分配给IO线程。

2. IO线程读取并解析请求

   主线程一旦把Socket分配给IO线程，就会进入阻塞状态，等待IO线程完成客户端请求读取和解析。因为有多个IO线程在并行处理，所以，这个过程很快就可以完成。

3. 主线程执行请求操作

   等到IO线程解析完请求，主线程还是会以单线程的方式执行这些命令操作

   

Redis 的数据结构？

答：

Redis 有 5 种 基本数据结构 ：String、List、Hash、Set（无序集合、值唯一）、Zset（有序集合）

有 3 种 特殊的数据结构 ：Geospatial、Hyperloglog、Bitmap

数据结构基本介绍：

• List 列表是链表，不是数据
• Set 集合内部的键值对是无序且唯一的

基本数据结构应用场景

String

最基础的数据结构，是二进制安全的、可以存储图片、可以存储序列化的对象，值最大存储 512M

底层使用 SDS 实现，与 C 语言的字符串底层实现不同，C 语言底层字符串使用 char[] 实现

String 的应用场景

• 限速器：防止 DoS 攻击，对 ip 进行访问次数限制，但是无法防止 DDoS 攻击，因为 DDoS 是分布式拒绝服务，使用了不同 ip 不断访问服务器

// 等价于 Set 192.168.55.1 ex 60 nx
// 如果该ip不存在，指定key为ip，value为1，过期时间为60秒
Boolean exists = Redis.Set(ip, 1, "ex 60", "nx");
if(exists != null || Redis.incr(ip) <= 5) {
  // 通过访问
} else {
  // 限流
}

• 共享 Session：在单体系统中，会使用 Session 保存用户的登陆信息，但是在分布式系统中，有多台应用，我们使用共享 Session 保证多台应用都有用户的登录信息，因此可以使用 Redis 存储用户的 Session，让多台应用可以来 Redis 中获取这个共享 Session

• 缓存对象：可以直接缓存对象的 JSON，Set user:1 '{"name": "zs", "age": 18}'

List

列表，用来存储多个值

List 的应用场景

• 栈 ：通过 lpush + lpop 实现
• 列表：通过 rpush + lpop 或 lpush + rpop 实现
• 阻塞式消息队列：通过 lpush + brpop 实现
• 秒杀中的异步队列，用来对高并发请求进行削峰

Hash

存储对象类型的数据：key 为对象名称，value 为描述对象属性的 Map，对象属性的修改在 Redis 中就可直接完成

Hash 的应用场景

• 缓存用户信息

Set

存储多个元素，不允许存在重复元素

Set 的应用场景

• 去重操作

• 用户标签

Zset

存储多个有序元素，不允许存在重复元素

Zset 的应用场景

• 用户排行榜
• 用户点赞统计

特殊的数据结构

这里再简单说一下 Redis 中 3 种特殊的数据结构的作用：

• Geospatial：用于存储地理位置信息
• Hyperloglog：用来做基数统计算法的数据结构，如统计网站的 UV。
• Bitmap：用一个比特位来映射某个元素的状态，在 Redis 中，它的底层是基于字符串类型实现的 。一般使用 Bitmap 来标记某些元素是否存在

Redis 基本数据结构底层实现

String 底层实现为 SDS 数据结构

List 底层实现为 双向链表 或 压缩列表 ，Redis3.2 之后 List 底层由 quicklist 实现

Hash 底层实现为 压缩列表 或 哈希表 ：哈希元素少于 512 个，并且每个元素的值都小于 64B，则会使用 压缩列表 ，否额会使用 哈希表

Set 底层实现为 哈希表 或 整数集合 ：集合元素少于 512 个，则会使用 整数集合 ，否额会使用 哈希表

Zset 底层实现为 压缩列表 或 跳表 ，集合元素少于 128 个，并且每个元素的值都小于 64B，则会使用 压缩列表 ，否则会使用 跳表

为了更好的性能，在 Redis7.0 中，将 压缩列表（zipList） 全部替换为了 Listpack，但是为了兼容，还是保留了 zipList 的相关属性

SDS 了解吗？

答：

Redis 创建了 SDS（simple dynamic String） 的抽象类型作为 String 的默认实现

SDS 的结构如下：

struct sdshdr {
  // 字节数组，用于保存字符串
  char buf[];
  // buf[]中已使用字节数量，称为SDS的长度
  int len;
  //  buf[]中尚未使用的字节数量
  int free;
}

为什么 Redis 不使用 C 语言默认的字符串呢？

• 提升获取长度性能：使用 SDS 可以以 O(1) 的时间复杂度取到字符串的长度，因为 SDS 中存储了字符串的长度；在 C 语言的字符串中，需要遍历才可以获取长度

• 保障二进制安全：C 字符串只能包含符合某种编码格式的字符，如 ASCII、UTF-8，并且除了字符串末尾外，不能含有 '\0'，这是因为 C 字符串是以 '\0' 结尾的，而在视频、图片中的数据以 '\0' 作为分隔符很常见，因此 C 字符串无法存储这些数据；而在 SDS 中存储了字符串的长度，因此不需要分隔符

• 减少内存再分配数：SDS 采用了空间预分配策略，每次 SDS 进行空间扩展时，程序会同时分配所需空间和额外的未使用空间，以减少内存的再分配次数。额外分配的未使用空间大小取决于空间扩展后SDS的len属性值。

  • 如果len < 1M，那么分配的未使用空间大小与 len 相同
  • 如果len >= 1M，那么分配的未使用空间大小为 1M

  SDS 也采用了惰性空间释放策略，即 SDS 字符串长度变短，并不会立即释放空间，而是将未使用的空间添加到 free 中，以便后期扩展 SDS 时减少内存分配次数

Redis 的 zipList

答：

Redis 的压缩列表即 zipList，可以包含多个节点，每个节点可以保存一个长度受限的字符数组或者整数 （Redis7.0 之后被废弃）

因为 zipList 节约内存的性质，哈希键、列表键和有序集合键初始化的底层实现皆采用 zipList

zipList 底层结构由 3 部分组成：head、entries、end，这三部分在内存上连续存放

• head

  head由三部分组成

  • zlbytes：占4B，表示 zipList 整体所占字节数，包括 zlbytes 本身长度
  • zltail：占4B，用于存放 zipList 最后一个 entry 在整个数据结构中的偏移量，可用于定位链表末尾的 entry
  • zllen：占2B，存放列表包含的 entry 个数

• entries

  entries 是真正的列表，由很多 entry 构成，由于元素类型、数值不同，每个 entry 的长度也不同

  entry由三部分组成

  • prevlength：记录上一个 entry 的长度，用于逆序遍历，默认长度为 1 字节，如果上一个 entry 的长度 >= 254B，prevlength 就会扩展为 5B
  • encoding：标志后面的 data 的具体类型。如果 data 为整数，encoding 固定长度为1B，如果data为字符串，encoding可能为1B、2B、5B，data字符串不同的长度，对应着不同的encoding长度
  • data：真正存储的数据，数据类型只能是整数或字符串

• end

  end只包含一部分，称为 zlend，占1B，值为255，也就是8个1，表示 zipList 列表的结束

Redis 的 ListPack

答：

Redis 的 zipList 存在一些缺点：

• 实现复杂，为了实现逆序遍历，每一个 entry 都存储了前一个 entry 的长度，这样在插入和更新时可能会造成连锁更新

  连锁更新举例： 假如 zipList 中每一个 entry 都是很接近但又不到 254B，此时每个 entry 的 prevlength 使用 1 个字节就可以保存上个节点的长度，但是此时如果向 zipList 中间插入一个新的节点，长度大于 254B，那么新插入节点后边的节点需要把 prevlength 扩展为 5B 来存储新插入节点的长度，那么扩展后该节点长度又大于 254B，因此后边节点需要再次扩展 prevlength 来存储该节点的长度，导致了插入节点后边的所有节点都需要更新 prevlength 的值，这属于是极端情况下才会发生

因此为了更好的性能，在 Redis7.0 中，将 zipList 全部替换为了 Listpack，但是为了兼容，还是保留了 zipList 的相关属性

• head

  head包括两部分：

  • totalBytes：占4B，存放 ListPack 整体所占字节数，包含 totalBytes 本身
  • elemNum：占 2B，存放 entry 个数

• entries

  entries是真正的列表，由很多entry组成，每个entry长度不同

  entry包括三部分（删除了prevlength，增加了element-total-len）：

  • encoding：标志后面的 data 的具体类型。如果 data 为整数，encoding 固定长度为 1B，如果 data 为字符串，encoding 可能为 1B、2B 或 5B，data 字符串不同的长度，对应着不同的 encoding 长度
  • data：存储真正数据
  • element-total-len：记录当前 entry 长度，用于实现逆序遍历，占1、2、3、4或5字节

Zset 的底层实现？为什么不用红黑树？

答：

Zset 的底层实现是：压缩列表 + 跳表

什么时候使用压缩列表？

• 有序集合保存的元素个数小于 128 个
• 有序集合保存的所有元素成员的长度都必须小于 64 字节

否则使用跳表

跳表在 Redis 中的作用就是作为有序集合类型的底层数据结构，

跳表中每个节点保存着其他节点的指针，高层的指针越过的元素数量大于等于低层的指针，因此在跳表中，在查找元素时可以一次跳过多个节点，当找到大于或等于目标元素的节点后，再使用普通指针开始移动（可以向后移动，也可以向前移动，跳表含有前边节点的指针）寻找目标元素，跳表可以在 O(logn) 的时间内遍历跳表

跳表结构图：

扩展：Zset 底层为什么不用红黑树？

跳表和红黑树的查找时间复杂度都是O(logn)，但是红黑树的插入/删除效率比较低 ，因为红黑树在插入删除时，还需要调整结构来保证红黑树的平衡

谈谈Redis 的持久化策略？

参考文章：

• Redis 持久化机制演进与百度智能云的实践 (opens new window)

Redis的确是将数据存储在内存的，但是也会有相关的持久化机制将内存持久化备份到磁盘，以便于重启时数据能够重新恢复到内存中，避免数据丢失的风险。而Redis持久化机制有三种：AOF、RDB、混合型持久化（4.x版本后提供）

RDB持久化

如果需要关闭 RDB 持久化只要将 save 的保存策略注释掉即可

RDB 是通过 快照的方式 实现持久化，根据条件判断是否出发 RDB 持久化，如果触发，则将此时内存中的数据快照写入到磁盘中，以二进制的压缩文件进行存储

RDB持久化的方式有两种：

• 手动触发（分为手动 save 和手动 bgsave）

  • 手动save：阻塞当前 Redis，直到持久化完成，可能会对客户端命令的执行造成长时间阻塞，线上不建议使用。

  • 手动bgsave：Redis 进程执行 fork 创建子进程进行持久化，阻塞时间很短。

    在执行Redis-cli shutdown关闭Redis服务时或执行flushall命令时，如果没有开启AOF持久化，会自动执行bgsave

  • 被动触发（以下四种情况会被动触发）

    • 达到了在 Redis.conf 中配置被动触发的条件，会触发 bgsave 生成 rdb 文件

      Redis 中 save 操作的配置：从右向左条件主键变弱，如果60s发生了10000次写操作，就进行持久化，如果没有达到，在300s时，如果有100次写操作就会持久化，如果没有达到在3600s，如果有一次写操作就会持久化

      

    • 主从复制时，从节点需要全量同步主节点的数据，会触发 bgsave

    • 执行 debug reload 命令重新加载 Redis 时，会触发 bgsave

    • 执行 shutdown 命令时，如果没有开启 aof 持久化，会触发 bgsave

• bgsave子进程工作原理：

由子进程继承父进程所有资源，且父进程不能拒绝子进程继承，bgsave子进程先将内存中的全量数据copy到磁盘的一个RDB临时文件，持久化完成后将该临时文件替换原来的dump.rdb文件。

如果持久化过程中出现了新的写请求，那么 Redis 并不会直接修改现有内存中的数据，而是将这块内存数据拷贝出来，再修改这块新内存的数据，这就是 写时复制（Copy on Write）

​

• 操作系统中的写时复制技术：

写时复制的优点在于：只有读请求的话，那么 fork 出来的子进程和父进程共享同一个内存区域，如果有写请求的话，就将需要写的数据所在的内存数据复制一份，写到新复制出来的内存数据块中（在读多写少时，避免了多次的内存拷贝）

在Linux系统中，调用 fork 系统调用创建子进程时，并不会把父进程所有占用的内存页复制一份，而是与父进程共用相同的内存页，而当子进程或者父进程对内存页进行修改时才会进行复制 —— 这就是著名的 写时复制 机制。

那么bgsave中的写时复制技术即如果在持久化过程中，写入了新的数据，此时再去将元数据重新拷贝一份，进行修改。

​

优点：

• 使用单独子进程持久化，保证 Redis 高性能。
• RDB 持久化存储压缩的二进制文件，适用于备份、全量复制，可用于灾难备份，同时RDB文件的加载速度远超于AOF文件。

缺点：

• 没有实时持久化，可能造成数据丢失。
• 备份时占用内存，因为Redis 在备份时会独立创建一个子进程，将数据写入到一个临时文件（需要的内存是原本的两倍）
• RDB文件保存的二进制文件存在新老版本不兼容的问题。

AOF持久化

AOF 通过 命令追加的方式 对数据进行持久化，相比于 RDB 方式持久化的更加及时

默认AOF没有开启，可在Redis.conf中配置

Redis7发生了重大变化，原来只有一个appendonly.aof文件，现在具有了三类文件：

• 基本文件：RDB 格式或 AOF 格式。存放 RDB 转为 AOF 时的内存快照数据。该文件可以有多个。
• 增量文件：以操作日志形式记录转为 AOF 后的写入操作。该文件可以有多个。
• 清单文件：维护 AOF 文件的创建顺序，保证激活时的应用顺序。该文件只可以有1个。

AOF 文件中存储的 resp 协议数据格式，如果执行命令Set a hello，aof文件内容如下：（*3代表有3条命令，$5代表有5个字符）

*3
$3
Set
$1
a
$5
hello

AOF持久化时，其实是先写入内存中，之后再同步到磁盘中，同步策略有三种：

• appendfsync always：每次写入都同步到磁盘，最安全，但影响性能。
• appendfsync everysec（推荐、默认配置）：每秒同步一次，最多丢失1秒的数据。
• appendfsync no ：Redis并不直接调用文件同步，而是交给操作系统来处理，操作系统可以根据buffer填充情况/通道空闲时间等择机触发同步；这是一种普通的文件操作方式。性能较好，在物理服务器故障时，数据丢失量会因OS配置有关。

AOF 优点：

• 数据丢失风险较低，后台线程处理持久化，不影响客户端请求处理的线程。

AOF 缺点：

• 文件体积由于保存的是所有命令会比RDB大上很多，而且数据恢复时也需要重新执行指令，在重启时恢复数据的时间往往会慢很多。

AOF的重写（Rewrite）机制：

• 为了防止 AOF 文件太大占用大量磁盘空间，降低性能，Redis 引入了 Rewrite 机制 对 AOF 文件进行压缩

  Rewrite 就是对AOF文件进行重写整理。当开启Rewrite，主进程Redis-server创建出一个子进程bgrewriteaof，由该子进程完成rewrite过程。

  首先会对现有aof文件进行重写，将计算结果写到一个临时文件，写入完毕后，再重命名为原aof文件，进行覆盖。

  为什么重写可以减小 AOF 文件大小呢？

  使用重写之后，就会将最新数据记录到 AOF 文件中，比如之前对于 name 属性设置了好多次，AOF 文件中记录了 set name n1，set name n2 ...，那么在 重写 之后，AOF 文件中就直接记录了 set name nn，nn 就是 name 的最新值，通过这样来减小 AOF 文件体积是

• 配置AOF重写频率

# auto‐aof‐rewrite‐min‐size 64mb //aof文件至少要达到64M才会自动重写，文件太小恢复速度本来就很快，重写的意义不大
# auto‐aof‐rewrite‐percentage 100 //aof文件自上一次重写后文件大小增长了100%则再次触发重写

• AOF的持久化流程图：

• 混合持久化开启

  默认开启，即AOF持久化的基本文件时的基本文件是RDB格式的。（必须先开启aof）

  

  混合持久化重写aof文件流程：aof 在重写时，不再将内存数据转为 resp 数据写入 aof 文件，而是将之前的内存数据做 RDB 快照处理，将 RDB快照+AOF增量数据 存在一起写入新的 AOF 文件，完成后覆盖原有的 AOF 文件。

  Redis重启加载数据流程：

  1. 先加载 RDB 数据到内存中
  2. 再重放增量 AOF 日志，加载 AOF 增量数据

  优点：

  • 结合了 RDB 和 AOF，既保证了重启 Redis 的性能，又降低数据丢失风险

  缺点：

  • AOF 文件中添加了 RDB 格式的内容，使得 AOF 文件的可读性变得很差；

Redis 的数据备份策略

答：

1、写crontab定时调度脚本，每小时都copy一份rdb或aof的备份到一个目录中去，仅仅保留最近48小时的备份

2、每天都保留一份当日的数据备份到一个目录中去，可以保留最近1个月的备份

3、每次copy备份的时候，都把太旧的备份给删了

4、每天晚上将当前机器上的备份复制一份到其他机器上，以防机器损坏

Redis的内存如果满了，会发生什么？

Redis 的内存如果达到阈值，就会触发 内存淘汰机制 来选择一些数据进行淘汰，如果淘汰之后还没有内存的话，就会返回写操作 error 提示（Redis 内存阈值通过 redis.conf 的 maxmemory 参数来设置）

Redis 中提供了 8 种内存淘汰策略：

• noeviction （默认策略）：不删除键，返回错误 OOM ，只能读取不能写入

• volatile-lru ：针对设置了过期时间的 key，使用 LRU 算法进行淘汰

• allkeys-lru ：针对所有 key 使用 LRU 算法进行淘汰

• volatile-lfu ：针对设置了过期时间的 key，使用 LFU 算法进行淘汰

• allkeys-lfu ：针对所有 key 使用LFU 算法进行淘汰

• volatile-random :从设置了过期时间的 key 中随机删除

• allkeys-random : 从所有 key 中随机删除

• volatile-ttl ：删除生存时间最近的一个键

Redis如何实现事务？

Redis 自身提供了 事务功能 ，但是并没有 回滚机制 ，Redis 的事务可以顺序执行队列中的命令，保证其他客户端提交的命令不会插入到事务执行的命令序列中

Redis 提供了 MULTI、EXEC、DISCARD 和 WATCH 这些命令来支持事务

Redis 事务的缺点：

• 不保证原子性： 事务执行过程中，如果所有命令入队时未报错，但是在事务提交之后，在执行的时候报错了，此时正确的命令仍然可以正常执行，因此 Redis 事务在该情况下不保证原子性
• 事务中的每个命令都需要与 Redis 进行网络通信

因此 Redis 自身的事务使用的比较少，而是更多的使用 Lua 脚本 来保证命令执行原子性，使用 Lua 脚本的 好处 ：

• 减少网络开销： 多个请求以脚本的形式通过一次网络 IO 即可发送到 Redis
• 原子操作： Redis 会原子性执行整个 Lua 脚本
• 复用： 客户端的 Lua 脚本会永久存在Redis，之后可以复用

这里 Redis Lua 脚本的原子性指的是保证在执行 Lua 脚本的时候，不会被其他操作打断，从而保证了原子性，但是在 Lua 脚本中如果发生了异常，异常前的命令还是会被正常执行，并且无法进行回滚， 因此要注意 Lua 中保证的原子性是指在 Lua 脚本执行过程中不会被其他操作打断

Lua 脚本在执行过程中不会被打断，因此注意不要在 Lua 脚本中执行比较耗时的操作，导致 Redis 阻塞时间过长！

接下来还问了有 Redis 如何实现滑动窗口限流、常见的限流算法、漏桶和令牌桶限流有什么区别、如何进行选择，这一块限流的内容放在下一篇文章说明！

如何用 Redis 实现滑动窗口限流？

这里基于 Redis 实现一个滑动窗口限流算法

• 什么是滑动窗口限流?

滑动窗口限流指在一定时间窗口内限制请求的数量，并且随着时间的推移，新增的请求会被记入新的时间段内，通过滑动窗口限流可以使得限流更加平滑

• 如何基于 Redis 实现滑动窗口限流？

可以基于 Redis 的 Zset 数据结构实现，每当有一个请求进来时，将当前请求加入到 Zset 中，并且通过 zremrangebyscore 将当前窗口之前的请求给移除掉

移除之后，此时 Zset 中的请求就是当前时间窗口中的请求了，此时通过 zcard 命令查询 Zset 集合中的请求数量，如果大于请求阈值，就拒绝之后的请求；否则，就可以接受之后的请求，使用的命令如下：

# 将当前请求加入到 slideWindows 滑动窗口限流中，将当前时间作为 score 加入
zadd slideWindows [currentTime] [currentTime] 
# 移除当前窗口之前的请求，currentTime - windowsDuartion * 1000 为当前时间窗口开始节点，windowsDuration 单位这里是 s，因此乘 1000 转为毫秒
zremrangebyscore slideWindows 0 currentTime - windowsDuartion * 1000
# 统计当前窗口请求数量
zcard slideWindows

Java 中判断逻辑如下（伪代码）：

// 最后判断窗口内的请求数量是否大于阈值
if (redisCache.zcard("slideWindows") >= WINDOWS_THRESHOLD) {
    // 达到阈值，返回 true，表示触发限流
    return true;
}
// 未达到阈值，返回 false，表示不触发限流
return false;

还了解哪些限流算法？

除了滑动窗口限流之外，还有两种常用的限流算法： 漏桶算法 和 令牌桶算法

Google Guava 包下的 RateLimiter 是基于令牌桶算法的思想实现，这里说一下这两种限流算法以及它们的区别

漏桶算法

漏桶算法的原理就是 将请求加入漏桶中，漏桶以固定速率出水，如果请求在漏桶中溢出就拒绝请求

那么这个漏桶就可以使用一定长度的队列来实现，长度就是这个漏桶所能容纳请求的数量，再通过另一个线程从队列的另一端去不断取出任务执行就可以了

• 漏桶算法存在的问题

漏桶算法存在的问题就是只能以固定速率处理到来的请求，无法处理突发请求 ，也就是一瞬间如果有超过漏桶大小的请求数量过来的话，超出的那部分请求就会被无情的抛弃

那么漏桶算法的这个问题在令牌桶算法中得到了解决 ，如果请求一开始数量较少，令牌桶中会积累令牌数量，当有突发流量到来的时候，会去使用已经积累的令牌数量来去处理这些请求，并且 RateLimiter 的实现中 还可以对未来令牌数量透支 ，这样 RateLimiter 实现的令牌桶算法就可以很好的应对突发流量了，不过这样带来的缺点就是如果一直并发量比较高，导致对未来的令牌数量一直透支，会导致后边请求的阻塞等待时间逐渐变长，解决方法就是适当的加一些请求拒绝策略就可以缓解这种现象

在高并发的场景中，突发流量还是比较常见的，因此在 RateLimiter 基于令牌桶算法实现中为了应对突发流量，做出了透支令牌的优化

漏桶算法如下图所示：

漏桶算法和令牌桶算法还有一点区别就是：

漏桶算法是需要将请求给存储在队列中，而在令牌桶算法中，并没有真正去产生令牌，而是根据时间差来计算这段时间应该产生的令牌数， 所以令牌桶算法的性能相对于漏桶算法来说是比较高的！

令牌桶算法

令牌桶算法的原理就是 系统使用恒定速率往桶中放入令牌，如果请求需要被处理，就从桶中获取令牌，如果没有令牌的话，请求被拒绝

RateLimiter 就是基于令牌桶算法实现的，在他里边并没有真正的去创建令牌实体，而是根据时间差来计算这一段时间产生的令牌数，这样做的好处就是 性能比较高

如果真正要去创建令牌实体的话，肯定需要再启动一个任务，以固定速率向令牌桶中生成令牌，那么启动一个新的任务是会带来一定的系统开销的，可能会加重系统的负担，那么通过时间差来计算令牌数的话，通过简单的计算就可以拿到产生的令牌数量，开销大大减少

两种算法的区别

漏桶算法 ：主要用于平滑流量，对于突发流量的应对不好，如果突发流量过大超出了队列长度就会被无情抛弃；并且需要将请求存储在队列中

令牌桶算法 ：为了更好应对突发流量引入了透支令牌的优化，但是如果一直透支对后来的请求也很不友好，有利有弊；并且 RateLimiter 中对令牌桶算法还做出了优化，并不真正去生成令牌实体，而是根据时间去计算应该生成的令牌数，降低系统开销

两种算法该如何选择呢？

在系统中，一般来说突发流量会比较多一些，因此限流算法如果应对突发流量更好一些的话，用户体验会更好，因此可以选择 令牌桶算法 ，基于令牌的方式，可以很好应对突发流量，并且还可以对未来令牌进行透支，不过透支未来令牌会导致后边的请求阻塞时间过长，因此可以考虑加入一定的拒绝策略，不要透支太多的令牌（详细见下方 RateLimiter 缺陷）

并且令牌桶算法基于时间来计算令牌，不需要生成令牌实体，而漏桶算法还要生成请求存储在队列中，因此令牌桶算法性能相对来说更好

扩展：RateLimiter 的缺陷

RateLimiter 是存在缺陷的，如果系统的并发量逐步升高，通过 acquire() 方法是一定会去获取令牌的，而由于 RateLimiter 中 透支未来令牌 的设计，这就会导致后边的请求等待时间会逐步升高，下边代码模拟了并发量逐步升高的场景，从输出结果看可以发现后边的请求等待的时间越来越长，这显然对后来的请求很不友好

public static void main(String[] args) throws InterruptedException {
    RateLimiter rateLimiter = RateLimiter.create(2);
    for (int i = 1; i < 20; i ++) {
        double acquire = rateLimiter.acquire(i);
        System.out.println("获取了" + i + "个令牌，等待时间为" + acquire);
    }
    /**
     * 输出：
     * 获取了1个令牌，等待时间为0.0
     * 获取了2个令牌，等待时间为0.499337
     * 获取了3个令牌，等待时间为0.998667
     * 获取了4个令牌，等待时间为1.499843
     * 获取了5个令牌，等待时间为1.996169
     * 获取了6个令牌，等待时间为2.499906
     * 获取了7个令牌，等待时间为2.993976
     * 获取了8个令牌，等待时间为3.499379
     * 获取了9个令牌，等待时间为3.999501
     * 获取了10个令牌，等待时间为4.490265
     */
}

• 怎么来解决这个问题呢？

这个问题的原因就是 acquire() 方法一定会获取令牌，那么我们在获取令牌之前可以先使用 tryAcquired 检测：

1、如果可行再去 acquire()

2、如果令牌不足，适当拒绝请求

因此解决策略就是我们去 定义一个拒绝策略 ，当发现等待的时间远远超出了可以接受的范围，就将该请求给拒绝掉，这样就不会导致一致透支未来的令牌，导致后边的请求越来越慢

• acquire 包装代码解析

如下代码（来源于 xjjdog 作者的 Github），我们将 acquire 方法给包装一下，先通过 tryAcquire() 尝试获取令牌，如果获取不到返回 false，我们再将请求数量给记录到原子类中，再通过 acquire() 开始阻塞等待获取令牌，当发现等待的请求数量超过指定的最大请求数量之后，就将之后的请求给拒绝掉！

public class FollowController {

    private final RateLimiter rateLimiter;

    private int maxPermits;

    private Object mutex = new Object();

    //等待获取permits的请求个数，原则上可以通过maxPermits推算
    private int maxWaitingRequests;

    private AtomicInteger waitingRequests = new AtomicInteger(0);

    public FollowController(int maxPermits,int maxWaitingRequests) {
        this.maxPermits = maxPermits;
        this.maxWaitingRequests = maxWaitingRequests;
        rateLimiter = RateLimiter.create(maxPermits);
    }

    public FollowController(int permits,long warmUpPeriodAsSecond,int maxWaitingRequests) {
        this.maxPermits = maxPermits;
        this.maxWaitingRequests = maxWaitingRequests;
        rateLimiter = RateLimiter.create(permits,warmUpPeriodAsSecond, TimeUnit.SECONDS);
    }

    public boolean acquire() {
        return acquire(1);
    }

    public boolean acquire(int permits) {
        boolean success = rateLimiter.tryAcquire(permits);
        if (success) {
            rateLimiter.acquire(permits);//可能有出入
            return true;
        }
        if (waitingRequests.get() > maxWaitingRequests) {
            return false;
        }
        waitingRequests.getAndAdd(permits);
        rateLimiter.acquire(permits);

        waitingRequests.getAndAdd(0 - permits);
        return true;
    }

}

缓存雪崩是什么？如何解决？

缓存雪崩造成的原因是：大量缓存数据在同一时间过期或者Redis宕机，此时如果有大量的请求无法在 Redis 中处理，会直接访问数据库，从而导致数据库的压力骤增，甚至数据库宕机

缓存过期解决：

1. 给过期时间加上一个随机数
2. 互斥锁，当缓存失效时，加互斥锁，保证同一时间只有一个请求来构建缓存
3. 缓存预热，在系统启动前，提前将热点数据加载到缓存中，避免大量请求同时访问数据库

Redis故障解决：

1. 服务熔断或请求限流
2. 构建 Redis 缓存高可靠集群

缓存穿透是什么？如何解决？

答：缓存穿透造成的原因是访问数据库中不存在的数据，即数据库和缓存都不命中

缓存穿透就是访问大量数据库中不存在的设备，每次都需要去数据库中查询，失去了缓存保护后端存储的意义。

造成原因：

• 自身代码问题
• 恶意攻击

解决方案有两种：

1、如果访问数据库中不存在的数据，则将该数据设置为字符串{}并且放入缓存，避免访问不存在的数据而大量请求打到数据库，缓存 kv 格式为：empty_cache_key: {}

// 设置key为空缓存
RedisUtil.Set(productCacheKey, "{}", 60 + new Random().nextInt(30), TimeUnit.SECONDS);
// 如果访问到空缓存，重新刷新空缓存的过期时间
RedisUtil.expire(productCacheKey, 60 + new Random().nextInt(30), TimeUnit.SECONDS);

2、布隆过滤器：在使用布隆过滤器时，先将所有数据Hash到一个位图中，之后接收客户端请求时，先去布隆过滤器中判断数据是否存在，如果不存在，则直接返回空，不会请求数据库。

在SpringBoot中，我们可以使用Guava提供的布隆过滤器实现缓存穿透的解决方案。

缓存击穿是什么？如何解决？

答：

缓存击穿造成的原因是：热点 key 失效

同一时间批量添加数据，并且数据的过期时间相同，大量数据同一时间缓存失效可能导致大量请求直达数据库，如果请求过多，数据库会挂掉。

解决：

批量添加数据的话，在设置的过期时间上再加上一个随机时间即可。

// 过期时间 = cache_timeout + new Random().nextInt(5) * 60 * 60;

还可通过互斥锁解决

Redis 的 key 删除策略了解吗？

答：

• 惰性删除：key过期后任然留在内存中不做处理，当有请求操作这个key的时候，会检查这个key是否过期，如果过期则删除，否则返回key对应的数据信息。（惰性删除对CPU是友好的，因为只有在读取的时候检测到过期了才会将其删除。但对内存是不友好，如果过期键后续不被访问，那么这些过期键将积累在缓存中，对内存消耗是比较大的。）
• 定期删除：Redis数据库默认每隔100ms就会进行随机抽取一些设置过期时间的key进行检测，过期则删除。（定期删除是定时删除和惰性删除的一个折中方案。可以根据实际场景自定义这个间隔时间，在CPU资源和内存资源上作出权衡。）
• Redis默认采用定期+惰性删除策略。

Redis 的集群架构

在生产环境中使用 Redis 一般不会使用单节点的方式部署，而是对 Redis 进行集群部署，来保证 Redis 的高可用

Redis 的集群部署方式有 3 种：主从复制、哨兵集群、切片集群部署

Redis 的主从复制

主从复制用于保证 Redis 服务的高可用，基本的主从复制集群就是： 主节点负责处理写请求，从节点负责处理读请求

主从复制好处

• 主从复制为后续的高可用机制打下了基础，可以将数据同步到多个从节点，做到灾备的效果
• 通过 主写从读 的形式实现读写分离，提高 Redis 吞吐量。

主从复制存在的问题：

• 如果主节点宕机，需要以手动的方式从 slave 选择一个新的 master，同时需修改应用方的主节点地址，还需要命令所有从节点去复制新的主节点，以手动的方式来设置新的 master 节点比较麻烦
• 由于只有主节点可以写入数据，因此写入性能受主节点的限制
• 木桶效应：整个Redis节点群能够存储的数据容量受到所有节点中内存最小的那台限制，比如一主两从架构：master=32GB、slave1=32GB、slave2=16GB，那么整个Redis节点群能够存储的最大容量为16GB

所以一般 Redis 集群部署不会仅仅使用主从复制的方式部署，而是会将主从复制和哨兵集群结合起来进行使用，通过哨兵来探测 master 节点的状态是否正常，来完成主从节点的自动切换

Redis 主从复制如何同步数据呢？

在 Redis2.8 之后，进行主从同步使用 psync 命令，在 2.8 之前使用 sync 命令，sync 在断线情况下会进行全量复制，效率很低，因此使用 psync 命令进行改进，具备了 全量同步 和 部分同步 的功能

psync 命令格式如下：

# runid 为 master 的身份 id
# offset 是从节点同步命令的偏移量
psync [runid] [offset]

那么在从节点第一次同步主节点数据时，会向主节点发送 psync ? -1 命令，那么 master 收到命令后，匹配 runid，如果匹配成功，会使用 bgsave 生成 RDB 文件快照，并将 RDB 文件发送给 slave，slave 在收到 RDB 快照后将数据载入

如果从节点是断线之后重新连接上主节点，那么在同步数据时，会向主节点发送 psync runid offset ，那么 master 在收到命令之后，如果 runid 匹配成功，会判断 offset 这个偏移量与 master 本机的数据缓存的偏移量相差是否超过了 复制积压缓冲区 的大小，如果超过了，说明 slave 断线时间太长了，master 的 复制积压缓冲区 中的数据已经和 slave 的数据不连续了，因此进行全量复制；否则，就进行增量复制，将 slave 断线期间没有收到的数据给发送一下就可以了

主节点在接收到命令时，如何保存命令并将命令增量复制给从节点？

master 在接收命令时，将命令传递给 slave 的同时，也会将命令存放到 复制积压缓冲区，并且记录当前积压队列中存放命令的偏移量 offset，当 slave 重连时，master 会根据 slave 传的 offset 和自己最新命令的 offset 进行比较，如果相差的大小超过 复制积压缓冲区 的大小，就直接进行全量复制；否则，就增量复制

复制积压缓冲区 其实就是一个环形的循环队列，默认大小为 1MB，该缓冲区大小越大，允许 slave 断线的时间就越长

#设置复制积压缓冲区大小
repl-backlog-size 1mb

Redis 4.0 PSYNC2.0

在 Redis2.8 之后，使用 psync runid offset 来实现增量同步，但是如果发生了 主从切换 ，那么新的 master 的 runid 和 offset 都会发生变化，导致还是需要全量复制

因此在 Redis4.0 的 PSYNC2.0 中优化了这个问题，即使发生了主从切换，如果条件允许，也可以进行增量同步

那么在 PSYNC2.0 中，舍弃了 runid 的概念，使用 replid 和 replid2 来代替：

• replid 表示自己的复制 id
• replid2 表示自己上一次复制的主库的 id，还使用 second_replid_offset 来记录上一次复制的 offset，replid2 和 second_replid_offset 是一对：

https://mp.weixin.qq.com/s?__biz=MjM5MDE0Mjc4MA%3D%3D&chksm=bdb8e58f8acf6c99f9f78a27b490281f19610a41fc6b34b9d8ed6fbc2448ee392a5a456913d6&idx=2&mid=2651135452&scene=27&sn=9c5c59d0119d4d01a6930d0adf2c90e4&utm_campaign=geek_search&utm_content=geek_search&utm_medium=geek_search&utm_source=geek_search&utm_term=geek_search#wechat_redirect

• 对 master 来说，replid 就是自己的复制 id，没有发生主从切换之前，replid2 为空，发生主从切换之后，新的 master 的 replid2 是旧 master 的 replid

• 对 slave 来说，replid 保存的是自己当前正在同步的 master 的 replid，replid2 保存的旧 master 的 replid

• second_replid_offset 记录了上一次复制的主库的 offset

psync2.0 中如何判断增量复制？

从节点向主节点发送同步请求，那么主节点需要判断两个条件：

1、从节点的 offset 是否在 复制积压缓冲区 中

2、判断复制历史是否一样

判断复制历史首先 master 要判断从节点传入的 replid 和 master 的 replid2 是否一致，如果一致（表明新的主节点和从节点之前都是复制的同一个主节点），再来判断这个 offset 是否在上一次的 second_replid_offset，如果也在的话，就可以进行增量同步

下边这个例子说明一下为什么 offset 要在上一次的 second_replid_offset 中：

比如一主一从机器来说，master1 和 slave1，master1 中写入两条数据：

# master1  
lpush A B
lpush A C

我们假设一条数据偏移量为 10，那么在 master1 中插入两条数据，master1 的 offset = 20，此时 master1 和 slave1 主从复制，slave1 在复制完第一条数据之后，也就是 lpush A B 之后，failover（故障恢复） 后 slave1 变为新的 master，我们先称原 slave1 为 master2，原 master1 为 slave2

那么 master2 又会接收新的请求，假如在 master2 中写入如下数据：

# master2
lpush A D

那么此时 master2 中也有两条数据了，如上图，在发生主从切换之后，master2 和 slave2 的 offset 都是 20，是相同的，但是他们两个能进行增量复制吗？

肯定不可以，虽然他们的 offset 是相同的，但是他的复制上下文已经变了，原来的 slave 进行主从复制是基于

lpush A B
lpush A C

这个上下文来进行复制的，那么在主从切换之后，新的主库写入了新的数据，新主库的上下文已经变为了：

lpush A B
lpush A D

和原来主库的上下文都不同了，那么只单纯比较 offset 是没有意义的，所以需要拿 offset 和 second_replid_offset 进行比较

PSYNC-AOF【扩展内容】

详细参考：Redis 主从复制演进历程与百度智能云的实践 (opens new window)

百度智能云提出了 - PSYNC-AOF 方案

在 Redis 的主从复制中存在的问题，首先就是 内存大小的限制 ，Redis 的数据都存储在内存中，Backlog（复制积压缓冲区） 也在内存中，那么内存容量是有限的，当主从连接断开重连后，从库在主库的 Backlog 中查找数据，因为 Backlog 容量是有限的，所以查找数据也有限，那么如果主从断连时间过长的话，就在 Backlog 中找不到对应的数据，就需要进行全量复制了

那么 PSYNC-AOF 方案中，将 AOF 的内容和 Backlog 内容保持一致，主从断连后，从库去主库的 Backlog 中查找数据，那么如果没有找到的话，尝试从 AOF 里找，那么通过 RDB 文件作为基准，AOF 作为增量，就可以实现一个更大的 复制积压缓冲区 ，从而可以应对更长的网络延迟以及网络断开。

总结：总之，主从复制的痛点在于如果主从连接断开时间过长，或者发生主从切换，会导致全量复制，全量复制很伤，所以优化主从数据同步的目的就是尽可能地避免全量复制

Redis Sentinel有什么作用？

答：

在主从集群基础上，使用 Sentinel（哨兵） 角色来帮我们监控 Redis 节点运行状态，并自动实现故障转移，当 master 节点出现故障时，Sentinel 根据规则选一个 slave 升级为 master，确保集群可用性，在这个过程中不需要人工介入。

Redis Sentinel的主要功能是：

• 监控 Redis 节点状态是否正常
• 故障转移，确保 Redis 系统的可用性

一般主从复制+哨兵一起使用，使用3台哨兵+1个主从集群（1master，2slave）

Redis Sentinel + 主从模式 存在的问题：

• Redis Sentinel 在主节点挂了之后，选举主节点中断时间达几秒甚至十几秒，期间无法进行写入操作
• 只有一个主节点对外提供服务，无法提供很高的并发，并且单个节点内存不宜设置过大，导致持久化文件过大，影响数据恢复和主从同步效率

哨兵集群如何判断主库下线？

当某个哨兵判断主库 主观下线 后，就会给其他哨兵发送命令，其他哨兵会根据自己和主库的连接情况，做出赞成或反对的响应，如果赞成数量大于 quorum 配置项，则判定主库客观下线了

quorum 在哨兵配置文件中进行配置，如果有 3 个哨兵节点，则 quorum 值默认为 2

主观下线：单个哨兵节点认为主库下线了

客观下线：大家都认为主库下线了

Redis 哨兵集群的选举

首先为什么需要哨兵集群？

因为哨兵也存在单点的情况，可能出现单点故障，为了保证哨兵集群的可用性，就需要哨兵的分布式集群，那么既然是分布式集群，就涉及到选举主节点的问题

那么哨兵集群的选举机制使用的就是 Raft 选举算法：当选举的票数大于等于 sum(哨兵节点数)/2 + 1 时，将成为主节点

哨兵选举为主节点需要满足条件：

• 拿到半数以上赞成票
• 拿到的票数还需要大于等于哨兵配置文件中的 quorum 值

哨兵集群选举新主库

当哨兵集群判断了主库客观下线 ，选择新的主库遵循以下规则：

• 选择从节点优先级最高的 slave-priority
• 选择复制偏移量最大、复制最完整的从节点

Redis 切片集群了解吗？

答：

Redis 切片集群是目前使用比较多的方案，Redis 切面集群支持多个主从集群进行横向扩容，架构如下：

使用切片集群有什么好处？

• 提升 Redis 读写性能 ：之前的主从模式中，只有 1 个 master 可以进行写操作，在切片集群中，多个 master 承担写操作压力

• 多个主从集群进行存储数据，比单个主从集群存储数据更多

  比如10G数据，1个主从集群的话，1个master需要存储10G，对有3个主从集群的切片集群来说，只需要master1存储3G，master2存储3G，master3存储4G即可

• 具备主从复制、故障转移

  切片集群中的每一个主从集群中，slave 节点不支持读，只做数据备份，因为已经有其他master节点分担压力

切片集群支持水平扩容，可以无限扩容吗？

不可以，官方推荐不要超过1000个，因为各个小集群之间需要互相进行通信，如果水平节点过多，会影响通信效率。

切片集群中插入数据时，数据被放在哪个master中？

Redis 切片集群中，数据是通过 哈希槽分区 来存储的，Redis 切片集群中有 16384 个哈希槽，每一个 master 会拿到一些槽位，在向切片集群中插入数据时，会根据 key 计算对应的哈希槽，插入到对应的哈希槽中，那么计算出来的哈希槽在哪个master中，数据当然也就被存放在对应的master上。（在切片集群中，只有master节点才有插槽，slave节点没有插槽）

故障恢复（Failover）：切片集群中如果一个master挂了，如何选举主节点？

当 master 挂了之后，该 master 下的所有 slave 会向所有节点广播 FAILOVER_AUTH_REQUEST 信息，其他节点收到后，只有 master 响应，master 会相应第一个收到的 FAILOVER_AUTH_REQUEST 信息，并返回一个 ack，尝试发送 failover 的 slave 会收集 master 返回的 FAILOVER_AUTH_ACK，当 slave 收到超过半数的 master 的 ack 后，就会变成新的 master。

这样会导致 slave 一直没收到超过半数的 master 的 ack，难道要一直选举吗？

其实不会导致 slave 一直选举的，因为在 slave 知道 master 挂了之后，会经过一个延时时间 delay 之后再去给所有节点发送选举消息

延迟时间计算：delay = 500ms + random(0~500ms) + slave_rank * 1000ms （版本不同可能不一样，原理大致相同）

slave_rank表示slave已经从master复制数据的总量的rank，rank越小，表示复制的数据越新，该slave节点也就越先发起选举。

因此数据量越多的 slave 就越早发送选举消息，也就越早得到 master 的 ack，成为新的 master 的可能性就越大

集群切片中的脑裂问题了解吗？如何解决？

比如一个主从集群：master1 对应两个从节点 slave1、slave2，如果master1和两个从节点出现网络分区，两个slave会选举出来一个新的master节点，客户端是可以感知到两个master，但是两个master之间因为网络分区无法感知，客户端会向这两个master都写入数据，之后如果网络分区恢复，其中一个master变为slave，就会导致数据丢失问题。

如何解决？添加Redis配置：

min-slave-to-write 1 

表示写数据时，写入master之后，不立即返回客户端写成功，而是去slave同步数据，取值为1表示最少同步1个slave之后，才算数据写成功，如果同步的slave节点数量没有达到我们配置的值，就算数据写失败，取值建议：集群总共3个节点，可以取1，这样集群中超过半数（1个master + 1个slave）都写入数据成功，才算写成功。

但是这个配置为了数据的一致性，牺牲了一定的集群可用性，如果一个master的所有slave都挂了，这个小集群就不可用了，master无法写入数据。

一般不使用，Redis丢一点数据也影响不大，所以主要还是保证Redis的可用性

集群是否完整才能对外提供服务？

当Redis.conf配置 cluster-require-full-coverage no 时，表示如果一个主从集群全部挂掉之后，集群仍然可用，如果为 yes，表示不可用

比如有3个主从集群，其中一个主从集群全部瘫痪，配置为no，则整个集群仍然可以正常工作

Redis集群如何对批量操作命令的支持？

对于 mSet、mget 这样的多个 key 的原生批量操作命令，Redis 集群只支持所有key落在同一个slot的情况，如果多个key一定要用mSet在Redis集群上操作，则可以在key之前加上{XX}，这样就会根据大括号中的内容计算分片Hash，确保不同的key落在同一slot内，例如：

mSet {user1}: name zhuge {user1}:age 18

虽然name和age计算出来的 Hash slot 值不同，但是前边加上大括号{user1}，Redis集群就会根据 user1 计算插槽值，最后name和age可以放入同一插槽。