基本数据类型

String

String 类型的底层的数据结构实现主要是 int 和 SDS（简单动态字符串）

  • SDS 不仅可以保存文本数据，还可以保存二进制数据。SDS 的所有 API 都会以处理二进制的方式来处理 SDS 存放在 buf[] 数组里的数据。所以 SDS 不光能存放文本数据，而且能保存图片、音频、视频、压缩文件这样的二进制数据
  • SDS 获取字符串长度的时间复杂度是 O(1) 因为 C 语言的字符串并不记录自身长度，所以获取长度的复杂度为 O(n)；而 SDS 结构里用 len 属性记录了字符串长度，所以复杂度为 O(1)

应用场景

（1）String常用来缓存对象,有以下两种方式：

直接缓存整个对象的JSON， SET user:1 '{"name":"xiaolin", "age":18}'

采用将 key 进行分离为 user:ID:属性，采用 MSET 存储，用 MGET 获取各属性值，命令例子： MSET user:1:name xiaolin user:1:age 18

（2）常规计数：因为 Redis 处理命令是单线程，所以执行命令的过程是原子的。因此 String 数据类型适合计数场景，比如计算访问次数、点赞、转发、库存数量

（3）分布式锁：SET 命令有个 NX 参数可以实现「key不存在才插入」，可以用它来实现分布式锁：

  • 如果 key 不存在，则显示插入成功，可以用来表示加锁成功；
  • 如果 key 存在，则会显示插入失败，可以用来表示加锁失败。

一般而言，还会对分布式锁加上过期时间，分布式锁的命令如下：

1
SET lock_key unique_value NX PX 10000
  • lock_key 就是 key 键；
  • unique_value 是客户端生成的唯一的标识；
  • NX 代表只在 lock_key 不存在时，才对 lock_key 进行设置操作；
  • PX 10000 表示设置 lock_key 的过期时间为 10s，这是为了避免客户端发生异常而无法释放锁。

而解锁的过程就是将 lock_key 键删除，但不能乱删，要保证执行操作的客户端就是加锁的客户端。所以解锁的时候，我们要先判断锁的 unique_value 是否为加锁客户端，是的话，才将 lock_key 键删除。

可以看到，解锁是有两个操作，这时就需要 Lua 脚本来保证解锁的原子性(参考Lua脚本实现分布式锁)，因为 Redis 在执行 Lua 脚本时，可以以原子性的方式执行，保证了锁释放操作的原子性：

1
2
3
4
5
6
7
8
// 释放锁时，先比较 unique_value 是否相等，避免锁的误释放
if redis.call("get",KEYS[1]) == ARGV[1] then
    return redis.call("del",KEYS[1])
else
    return 0
end

//lua脚本示例

这样一来，就通过使用 SET 命令和 Lua 脚本在 Redis 单节点上完成了分布式锁的加锁和解锁

(4) 共享 Session 信息: 通常我们在开发后台管理系统时，会使用 Session 来保存用户的会话(登录)状态，这些 Session 信息会被保存在服务器端，但这只适用于单系统应用，如果是分布式系统此模式将不再适用。例如用户一的 Session 信息被存储在服务器一，但第二次访问时用户一被分配到服务器二，这个时候服务器并没有用户一的 Session 信息，就会出现需要重复登录的问题，问题在于分布式系统每次会把请求随机分配到不同的服务器，因此，我们需要借助 Redis 对这些 Session 信息进行统一的存储和管理，这样无论请求发送到那台服务器，服务器都会去同一个 Redis 获取相关的 Session 信息，这样就解决了分布式系统下 Session 存储的问题

List

List 列表是简单的字符串列表，按照插入顺序排序，可以从头部或尾部向 List 列表添加元素,List 类型的底层数据结构是由双向链表或压缩列表实现的,但是在 Redis 3.2 版本之后，List 数据类型底层数据结构就只由 quicklist 实现了，替代了双向链表和压缩列表,常用命令如下：

1
2
3
4
5
6
# 将一个或多个值value插入到key列表的表头(最左边)，最后的值在最前面
LPUSH key value [value ...] 
# 将一个或多个值value插入到key列表的表尾(最右边)
RPUSH key value [value ...]
# 返回列表key中指定区间内的元素，区间以偏移量start和stop指定，从0开始
LRANGE key start stop

应用场景

（1）消息队列：消息队列在存取消息时，必须要满足三个需求，分别是消息保序、处理重复的消息和保证消息可靠性。Redis 的 List 和 Stream 两种数据类型，就可以满足消息队列的这三个需求。List 可以使用 LPUSH + RPOP （或者反过来，RPUSH+LPOP）命令实现消息队列：

不过，在消费者读取数据时，有一个潜在的性能风险点：
在生产者往 List 中写入数据时，List 并不会主动地通知消费者有新消息写入，如果消费者想要及时处理消息，就需要在程序中不停地调用 RPOP 命令（比如使用一个while(1)循环）。如果有新消息写入，RPOP命令就会返回结果，否则，RPOP命令返回空值，再继续循环。所以，即使没有新消息写入List，消费者也要不停地调用 RPOP 命令，这就会导致消费者程序的 CPU 一直消耗在执行 RPOP 命令上，带来不必要的性能损失，为了解决这个问题，Redis提供了 BRPOP 命令。BRPOP命令也称为阻塞式读取，客户端在没有读到队列数据时，自动阻塞，直到有新的数据写入队列，再开始读取新数据。和消费者程序自己不停地调用RPOP命令相比，这种方式能节省CPU开销

同时消费者要实现重复消息的判断，这需要2个方面的要求：

  • 每个消息都有一个全局的 ID
  • 消费者要记录已经处理过的消息的 ID。当收到一条消息后，消费者程序就可以对比收到的消息 ID 和记录的已处理过的消息 ID，来判断当前收到的消息有没有经过处理。如果已经处理过，那么，消费者程序就不再进行处理了

但是 List 并不会为每个消息生成 ID 号，所以我们需要自行为每个消息生成一个全局唯一ID，生成之后，我们在用 LPUSH 命令把消息插入 List 时，需要在消息中包含这个全局唯一 ID。例如，执行以下命令，就把一条全局 ID 为 111000102、库存量为 99 的消息插入了消息队列：

1
2
> LPUSH mq "111000102:stock:99"
(integer) 1

基于 List 类型的消息队列，满足消息队列的三大需求（消息保序、处理重复的消息和保证消息可靠性）

  • 消息保序：使用 LPUSH + RPOP；
  • 阻塞读取：使用 BRPOP；
  • 重复消息处理：生产者自行实现全局唯一 ID；
  • 消息的可靠性：使用 BRPOP、LPUSH

List 作为消息队列有什么缺陷？

List 不支持多个消费者消费同一条消息，因为一旦消费者拉取一条消息后，这条消息就从 List 中删除了，无法被其它消费者再次消费。要实现一条消息可以被多个消费者消费，那么就要将多个消费者组成一个消费组，使得多个消费者可以消费同一条消息(让一条消息转发至对应的组内，组内所有成员都可以收到消息)，但是 List 类型并不支持消费组的实现。

Hash

Hash 是一个键值对（key - value）集合，其中 value 的形式如： value=[{field1，value1}，...{fieldN，valueN}]。Hash 特别适合用于存储对象,Hash 与 String 对象的区别如下图所示:

常用命令如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
SET user '{"name":"harrykane", "age":29}'   #String类型 Json存储User对象
HMSET user harrykane name age 30            #Hash存储Player对象,key为实体类名，field-value为键值对

### 可以使用如下命令，将用户对象的信息存储到 Hash 类型
HMSET uid:1 name Tom age 15
HMSET uid:2 name Jerry age 13
HMSET uid:3 name David age 20

#获取uid为1对象操作如下：
HGETALL  uid:1
1) "name"
2) "Tom"
3) "age"
4) "15"

上面介绍到String + Json也是存储对象的一种方式，那么存储对象时，那么到底用 String + json 还是用 Hash 呢？实际上一般对象用String + Json 存储，对象中某些频繁变化的属性可以考虑抽出来用 Hash 类型存储。

Set&Zset

Set 类型是一个无序并唯一的键值集合，它的存储顺序不会按照插入的先后顺序进行存储,Set 类型和 List 类型的区别如下：

  • List 可以存储重复元素，Set 只能存储非重复元素；
  • List 是按照元素的先后顺序存储元素的，而 Set 则是无序方式存储元素的

Set 类型比较适合用来数据去重和保障数据的唯一性,例如点赞,Set 类型可以保证一个用户只能点一个赞；Set 类型同时支持交集运算，所以可以用来计算共同关注的好友、公众号等

Zset 类型（也称SortedSet）相比于 Set 类型多了一个排序属性 score（分值），对于有序集合 Zset 来说，每个存储元素相当于有两个值组成的，一个是有序集合的元素值，一个是排序值。有序集合保留了集合不能有重复成员的特性（分值可以重复），但不同的是有序集合中的元素可以排序:

Zset

Zset 类型（Sorted Set，有序集合） 可以根据元素的权重来排序，可以自己来决定每个元素的权重值。比如说，我们可以根据元素插入 Sorted Set 的时间确定权重值，先插入的元素权重小，后插入的元素权重大,在面对需要展示最新列表、排行榜等场景时，如果数据更新频繁或者需要分页显示，可以优先考虑使用 Sorted Set,有序集合比较典型的使用场景就是排行榜。例如学生成绩的排名榜、游戏积分排行榜、视频播放排名、电商系统中商品的销量排名等。

以博文点赞排名为例，小林发表了五篇博文，分别获得赞为 200、40、100、50、150

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# 往有序集合key中加入带分值元素
ZADD key score member [[score member]...] 

# arcticle:1 文章获得了200个赞
> ZADD user:xiaolin:ranking 200 arcticle:1
(integer) 1
# arcticle:2 文章获得了40个赞
> ZADD user:xiaolin:ranking 40 arcticle:2
(integer) 1
# arcticle:3 文章获得了100个赞
> ZADD user:xiaolin:ranking 100 arcticle:3
(integer) 1
# arcticle:4 文章获得了50个赞
> ZADD user:xiaolin:ranking 50 arcticle:4
(integer) 1
# arcticle:5 文章获得了150个赞
> ZADD user:xiaolin:ranking 150 arcticle:5
(integer) 1

获取小林文章赞数最多的 3 篇文章，可以使用 ZREVRANGE 命令（倒序获取有序集合 key 从start下标到stop下标的元素）：

1
2
3
4
5
6
7
8
# WITHSCORES 表示把 score 也显示出来
> ZREVRANGE user:xiaolin:ranking 0 2 WITHSCORES
1) "arcticle:1"
2) "200"
3) "arcticle:5"
4) "150"
5) "arcticle:3"
6) "100"

Stream

Redis Stream 是 Redis 5.0 版本新增加的数据类型，Redis 专门为消息队列设计的数据类型。

在 Redis 5.0 Stream 没出来之前，消息队列的实现方式都有着各自的缺陷，例如：

  • 发布订阅模式，不能持久化也就无法可靠的保存消息，并且对于离线重连的客户端不能读取历史消息的缺陷；
  • List 实现消息队列的方式不能重复消费，一个消息消费完就会被删除，而且生产者需要自行实现全局唯一 ID。

基于以上问题，Redis 5.0 便推出了 Stream 类型也是此版本最重要的功能，用于完美地实现消息队列，它支持消息的持久化、支持自动生成全局唯一 ID、支持 ack 确认消息的模式、支持消费组模式等，让消息队列更加的稳定和可靠

消息队列

Stream 可以以使用 XGROUP 创建消费组，创建消费组之后，Stream 可以使用 XREADGROUP 命令让消费组内的消费者读取消息,创建两个消费组，这两个消费组消费的消息队列是 mymq，都指定从第一条消息开始读取：

1
2
3
4
5
6
# 创建一个名为 group1 的消费组，0-0 表示从第一条消息开始读取。
> XGROUP CREATE mymq group1 0-0
OK
# 创建一个名为 group2 的消费组，0-0 表示从第一条消息开始读取。
> XGROUP CREATE mymq group2 0-0
OK

消息队列中的某一条消息一旦被消费组里的一个消费者读取了，就不能再被该消费组内的其他消费者读取了，即同一个消费组里的消费者不能消费同一条消息，但是，不同消费组的消费者可以消费同一条消息（但是有前提条件，创建消息组的时候，不同消费组指定了相同位置开始读取消息:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
127.0.0.1:6379> xadd  mymq *  name xiaolin
"1679378689141-0"
127.0.0.1:6379> XGROUP CREATE mymq group1 0-0
OK
127.0.0.1:6379>  XGROUP CREATE mymq group2 0-0
OK
127.0.0.1:6379> XREADGROUP GROUP group1 consumer1 STREAMS mymq >
1) 1) "mymq"
   2) 1) 1) "1679378689141-0"
         2) 1) "name"
            2) "xiaolin"
127.0.0.1:6379> XREADGROUP GROUP group1 consumer1 STREAMS mymq >
(nil)
127.0.0.1:6379> XREADGROUP GROUP group1 consumer2 STREAMS mymq >
(nil)
127.0.0.1:6379> XREADGROUP GROUP group2 consumer1 STREAMS mymq >
1) 1) "mymq"
   2) 1) 1) "1679378689141-0"
         2) 1) "name"
            2) "xiaolin"

使用消费组的目的是让组内的多个消费者共同分担读取消息，所以，我们通常会让每个消费者读取部分消息，从而实现消息读取负载在多个消费者间是均衡分布的，如下所示：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
##再次添加四条消息，这样mymq中一共有5条消息

127.0.0.1:6379> XADD mymq *  player kane
"1679378913264-0"
127.0.0.1:6379> XADD mymq *  age 30
"1679378922998-0"
127.0.0.1:6379> XADD mymq *  club  spurs
"1679378928773-0"
127.0.0.1:6379> XADD mymq *  number   9
"1679378936160-0"

##设置三个消费者组，观察其读取消息的情况：

127.0.0.1:6379>  XREADGROUP GROUP group2 consumer1 COUNT 1 STREAMS mymq >
1) 1) "mymq"
   2) 1) 1) "1679378913264-0"
         2) 1) "player"
            2) "kane"
127.0.0.1:6379>  XREADGROUP GROUP group1 consumer1 COUNT 1 STREAMS mymq >
1) 1) "mymq"
   2) 1) 1) "1679378913264-0"
         2) 1) "player"
            2) "kane"
127.0.0.1:6379>  XREADGROUP GROUP group3 consumer1 COUNT 1 STREAMS mymq >
1) 1) "mymq"
   2) 1) 1) "1679378689141-0"
         2) 1) "name"
            2) "xiaolin"
127.0.0.1:6379>  XREADGROUP GROUP group3 consumer2 COUNT 1 STREAMS mymq >
1) 1) "mymq"
   2) 1) 1) "1679378913264-0"
         2) 1) "player"
            2) "kane"
127.0.0.1:6379>  XREADGROUP GROUP group3 consumer3 COUNT 1 STREAMS mymq >
1) 1) "mymq"
   2) 1) 1) "1679378922998-0"
         2) 1) "age"
            2) "30"
127.0.0.1:6379>  XREADGROUP GROUP group3 consumer4 COUNT 1 STREAMS mymq >
1) 1) "mymq"
   2) 1) 1) "1679378928773-0"
         2) 1) "club"
            2) "spurs"
127.0.0.1:6379>  XREADGROUP GROUP group3 consumer5 COUNT 1 STREAMS mymq >
1) 1) "mymq"
   2) 1) 1) "1679378936160-0"
         2) 1) "number"
            2) "9"
127.0.0.1:6379>  XREADGROUP GROUP group3 consumer6 COUNT 1 STREAMS mymq >
(nil)
127.0.0.1:6379>

再次验证了同一条消息只能被一个消费者组里的一个消费者consumer所读取，但是可以被不同消费者组的多个消费者所读取，并且，一个队列中的所有消息会被一个消费者组里的所有消费者均匀分配（组内的多个消费者共同分担读取消息，所以，我们通常会让每个消费者读取部分消息，从而实现消息读取负载在多个消费者间是均衡分布的）

基于 Stream 实现的消息队列，如何保证消费者在发生故障或宕机再次重启后，仍然可以读取未处理完的消息？

Streams 会自动使用内部队列（也称为 PENDING List）留存消费组里每个消费者读取的消息，直到消费者使用 XACK 命令通知 Stream“消息已经处理完成”。消费确认增加了消息的可靠性，一般在业务处理完成之后，需要执行 XACK 确认消息已经被消费完成

Stream-Ack

如果消费者没有成功处理消息，它就不会给 Streams 发送 XACK 命令，消息仍然会留存。此时，消费者可以在重启后，用 XPENDING 命令查看已读取、但尚未确认处理完成的消息，查看 group2 中各个消费者已读取、但尚未确认的消息个数命令如下：

1
2
3
4
5
6
7
8
9
10
127.0.0.1:6379> XPENDING mymq group2
1) (integer) 3
2) "1654254953808-0"  # 表示 group2 中所有消费者读取的消息最小 ID
3) "1654256271337-0"  # 表示 group2 中所有消费者读取的消息最大 ID
4) 1) 1) "consumer1"
      2) "1"
   2) 1) "consumer2"
      2) "1"
   3) 1) "consumer3"
      2) "1"

Redis 基于 Stream 消息队列与专业的消息队列有哪些差距？

  1. Redis Stream 消息会丢失吗

  • Redis 生产者会不会丢消息？生产者会不会丢消息，取决于生产者对于异常情况的处理是否合理。 从消息被生产出来，然后提交给 MQ 的过程中，只要能正常收到 （ MQ 中间件） 的 ACK 确认响应，就表示发送成功，所以只要处理好返回值和异常，如果返回异常则进行消息重发，那么这个阶段是不会出现消息丢失的。

  • Redis 消费者会不会丢消息？不会，因为 Stream （ MQ 中间件）会自动使用内部队列（也称为 PENDING List）留存消费组里每个消费者读取的消息，但是未被确认的消息。消费者可以在重启后，用 XPENDING 命令查看已读取、但尚未确认处理完成的消息。等到消费者执行完业务逻辑后，再发送消费确认 XACK 命令，也能保证消息的不丢失。

  • Redis 消息中间件会不会丢消息？会，Redis 在以下 2 个场景下，都会导致数据丢失：

Redis 在队列中间件环节无法保证消息不丢。像 RabbitMQ 或 Kafka 这类专业的队列中间件，在使用时是部署一个集群，生产者在发布消息时，队列中间件通常会写「多个节点」，也就是有多个副本，这样一来，即便其中一个节点挂了，也能保证集群的数据不丢失

2、Redis Stream 消息可堆积吗？

Redis 的数据都存储在内存中，这就意味着一旦发生消息积压，则会导致 Redis 的内存持续增长，如果超过机器内存上限，就会面临被 OOM 的风险。所以 Redis 的 Stream 提供了可以指定队列最大长度的功能。当指定队列最大长度时，队列长度超过上限后，旧消息会被删除，只保留固定长度的新消息。这么来看，Stream 在消息积压时，如果指定了最大长度，还是有可能丢失消息的。但 Kafka、RabbitMQ 专业的消息队列它们的数据都是存储在磁盘上，当消息积压时，无非就是多占用一些磁盘空间

持久化

AOF

AOF重写

AOF 日志是一个文件，随着执行的写操作命令越来越多，文件的大小会越来越大。所以Redis 为了避免 AOF 文件越写越大，提供了 AOF 重写机制，当 AOF 文件的大小超过所设定的阈值后，Redis 就会启用 AOF 重写机制，来压缩 AOF 文件。AOF 重写机制是在重写时，读取当前数据库中的所有键值对，然后将每一个键值对用一条命令记录到「新的 AOF 文件」，等到全部记录完后，就将新的 AOF 文件替换掉现有的 AOF 文件：

在使用重写机制后，就会读取 name 最新的 value（键值对） ，然后用一条 「set name xiaolincoding」命令记录到新的 AOF 文件，之前的第一个命令就没有必要记录了，因为它属于「历史」命令，没有作用了。这样一来，一个键值对在重写日志中只用一条命令就行了。重写机制的妙处在于，尽管某个键值对被多条写命令反复修改，最终也只需要根据这个「键值对」当前的最新状态，然后用一条命令去记录键值对，代替之前记录这个键值对的多条命令，这样就减少了 AOF 文件中的命令数量

RDB

RDB 快照就是记录某一个瞬间的内存数据，记录的是实际数据，而 AOF 文件记录的是命令操作的日志，而不是实际的数据，因此在 Redis 恢复数据时， RDB 恢复数据的效率会比 AOF 高些，因为直接将 RDB 文件读入内存就可以，不需要像 AOF 那样还需要额外执行操作命令的步骤才能恢复数据，Redis 提供了两个命令来生成 RDB 文件，分别是 save 和 bgsave，他们的区别就在于是否在「主线程」里执行：

  • 执行了 save 命令，就会在主线程生成 RDB 文件，由于和执行操作命令在同一个线程，所以如果写入 RDB 文件的时间太长，会阻塞主线程；
  • 执行了 bgsave 命令，会创建一个子进程来生成 RDB 文件，这样可以避免主线程的阻塞；

在子进程执行 bgsave 过程中，Redis主线程依然可以继续处理操作命令的，也就是说数据是能被修改的，这依赖于写入时复制（Copy On Write）:

执行 bgsave 命令的时候，会通过 fork() 创建子进程，此时子进程和父进程是共享同一片内存数据的，由于共享父进程的所有内存数据，于是就可以直接读取主线程（父进程）里的内存数据，并将数据写入到 RDB 文件

  1. 当主线程（父进程）对这些共享的内存数据也都是只读操作，那么，主线程（父进程）和 bgsave 子进程相互不影响
  2. 但是，如果主线程（父进程）要修改共享数据里的某一块数据（比如键值对 A）时，就会发生写入时复制，于是这块数据的物理内存就会被复制一份（键值对 A'），然后主线程在这个数据副本（键值对 A'）进行修改操作。与此同时，bgsave 子进程可以继续把原来的数据（键值对 A）写入到 RDB 文件，这种情况下RDB 快照保存的是原本的内存数据，而主线程刚修改的数据，是没办法在这一时间写入 RDB 文件的，只能交由下一次的 bgsave 快照

混合持久化

将 RDB 和 AOF 合体使用，这个方法是在 Redis 4.0 提出的，该方法叫混合使用 AOF 日志和内存快照，也叫混合持久化。如果想要开启混合持久化功能，可以在 Redis 配置文件将下面这个配置项设置成 yes：

1
aof-use-rdb-preamble yes

使用了混合持久化后，AOF 文件的前半部分是 RDB 格式的全量数据，后半部分是 AOF 格式的增量数据，这样的好处在于重启 Redis 加载数据的时候，由于前半部分是 RDB 内容，这样加载的时候速度会很快。加载完 RDB 的内容后，才会加载后半部分的 AOF 内容，这里的内容是 Redis 后台子进程重写 AOF 期间，主线程处理的操作命令，可以使得数据更少的丢失

高可用

补充： Redis删除策略、内存淘汰策略

定时删除策略的做法是，在设置 key 的过期时间时，同时创建一个定时事件，当时间到达时，由事件处理器自动执行 key 的删除操作

惰性删除策略的做法是，不主动删除过期键，每次从数据库访问 key 时，都检测 key 是否过期，如果过期则删除该 key

定期删除策略的做法是，每隔一段时间「随机」从数据库中取出一定数量的 key 进行检查，并删除其中的过期key

Redis 使用的过期删除策略是「惰性删除+定期删除」，删除的对象是已过期的 key。

内存淘汰策略是解决内存过大的问题，当 Redis 的运行内存超过最大运行内存时，就会触发内存淘汰策略（前面说的过期删除策略，是删除已过期的 key，而当 Redis 的运行内存已经超过 Redis 设置的最大内存之后，则会使用内存淘汰策略删除符合条件的 key，以此来保障 Redis 高效的运行）

主从复制

主从复制模式下：主服务器可以进行读写操作，当发生写操作时自动将写操作同步给从服务器，而从服务器一般是只读，并接受主服务器同步过来写操作命令，然后执行这条命令

可以使用 replicaof（Redis 5.0 之前使用 slaveof）命令形成主服务器和从服务器的关系

主从服务器间的第一次同步的过程可分为三个阶段：

  • 第一阶段是建立连接、协商同步；
  • 第二阶段是主服务器同步数据给从服务器（主服务器会执行 bgsave 命令来生成 RDB 文件，然后把文件发送给从服务器）；
  • 第三阶段是主服务器发送新写操作命令给从服务器（会有数据一致性问题）

主从服务器在完成第一次同步后，双方之间就会维护一个 TCP 连接，后续主服务器可以通过这个连接继续将写操作命令传播给从服务器，然后从服务器执行该命令，使得与主服务器的数据库状态相同。而且这个连接是长连接的，目的是避免频繁的 TCP 连接和断开带来的性能开销。上面的这个过程被称为基于长连接的命令传播，通过这种方式来保证第一次同步后的主从服务器的数据一致性

总结

主从复制共有三种模式：全量复制、基于长连接的命令传播、增量复制

主从服务器第一次同步的时候，就是采用全量复制，此时主服务器会两个耗时的地方，分别是生成 RDB 文件和传输 RDB 文件。为了避免过多的从服务器和主服务器进行全量复制，可以把一部分从服务器升级为「经理角色」，让它也有自己的从服务器，通过这样可以分摊主服务器的压力

第一次同步完成后，主从服务器都会维护着一个长连接，主服务器在接收到写操作命令后，就会通过这个连接将写命令传播给从服务器，来保证主从服务器的数据一致性

如果遇到网络断开又恢复后，主服务器会采用增量复制的方式继续同步，也就是说只会把网络断开期间主服务器接收到的写操作命令，同步给从服务器

哨兵

哨兵（Sentinel）机制，它的作用是实现主从节点故障转移。它会监测主节点是否存活，如果发现主节点挂了，它就会选举一个从节点切换为主节点，并且把新主节点的相关信息通知给从节点和客户端，哨兵其实是一个运行在特殊模式下的 Redis 进程，所以它也是一个节点。从“哨兵”这个名字也可以看得出来，它相当于是“观察者节点”，观察的对象是主从节点

如何判断主节点真的故障了？

哨兵会每隔 1 秒给所有主从节点发送 PING 命令，当主从节点收到 PING 命令后，会发送一个响应命令给哨兵，这样就可以判断它们是否在正常运行，哨兵在部署的时候不会只部署一个节点，而是用多个节点部署成哨兵集群，当一个哨兵判断主节点为「主观下线」后，就会向其他哨兵发起命令，其他哨兵收到这个命令后，就会根据自身和主节点的网络状况，做出赞成投票或者拒绝投票的响应，当这个哨兵的赞同票数达到哨兵配置文件中的 quorum 配置项设定的值后，这时主节点就会被该哨兵标记为「客观下线」，并准备执行主从节点切换

执行切换之前还需要在哨兵集群中选出一个 leader（标记客观下线的哨兵向其他哨兵发起投票，询问是否可以当选leader），让 leader 来进行切换操作

主从故障转移操作包含以下四个步骤：

  • 第一步：在已下线主节点（旧主节点）属下的所有「从节点」里面，挑选出一个从节点，并将其转换为主节点。
  • 第二步：让已下线主节点属下的所有「从节点」修改复制目标，修改为复制「新主节点」；
  • 第三步：将新主节点的 IP 地址和信息，通过「发布者/订阅者机制」通知给客户端(每个哨兵节点提供发布者/订阅者机制，客户端可以从哨兵订阅消息)；
  • 第四步：继续监视旧主节点，当这个旧主节点重新上线时，将它设置为新主节点的从节点

集群

Redis 的哨兵模式基本已经可以实现高可用，读写分离 ，但是在这种模式下每台 Redis 服务器都存储相同的数据，很浪费内存，所以在 redis3.0 上加入了 Cluster 集群模式（Redis集群是去中心化的集群架构），同时实现了 Redis 的分布式存储（对数据进行分片，也就是说每台 Redis 节点上存储不同的内容）：

这里的 6 台 redis节点两两之间并不是独立的，每个节点都会通过集群总线(cluster bus)与其他的节点进行通信（Gossip协议），Redis 集群有16384 个哈希槽，每个 key 通过 CRC16 校验后对 16384 取模来决定放置哪个槽，集群的每个节点负责一部分hash槽，举个例子，比如当前集群有3个节点，那么：

  • 节点 A 包含 0 到 5460 号哈希槽
  • 节点 B 包含 5461 到 10922 号哈希槽
  • 节点 C 包含 10923 到 16383 号哈希槽

使用哈希槽的好处就在于可以方便的添加或者移除节点，并且无论是添加删除或者修改某一个节点，都不会造成集群不可用的状态。当需要增加节点时，只需要把其他节点的某些哈希槽挪到新节点就可以了；当需要移除节点时，只需要把移除节点上的哈希槽挪到其他节点就行了

gossip协议，是基于流行病传播方式的节点或者进程之间信息交换的协议。原理就是在不同的节点间不断地通信交换信息，一段时间后，所有的节点就都有了整个集群的完整信息，并且所有节点的状态都会达成一致。每个节点可能知道所有其他节点，也可能仅知道几个邻居节点，但只要这些节可以通过网络连通，最终他们的状态就会是一致的。Gossip协议最大的好处在于，即使集群节点的数量增加，每个节点的负载也不会增加很多，几乎是恒定的。

Redis集群采用无中心结构,它的特点如下：

  1. 所有的redis节点彼此互联(PING-PONG机制),内部使用二进制协议优化传输速度和带宽
  2. 节点的fail是通过集群中超过半数的节点检测失效时才生效
  3. 客户端与redis节点直连,不需要中间代理层.客户端不需要连接集群所有节点,连接集群中任何一个可用节点即可

故障转移

Redis集群通过ping/pong消息，实现故障发现。这个环境包括主观下线和客观下线：

主观下线： 某个节点认为另一个节点不可用，即下线状态，这个状态并不是终的故障判定，只能代表一个节点的意见，可能存在误判情况

客观下线： 指标记一个节点真正的下线，集群内多个节点都认为该节点不可用，从而达成共识的结果。如果是持有槽的主节点故障，需要为该节点进行故障转移（master-slave切换）

故障发现后，如果下线节点的是主节点，则需要在它的从节点中选一个替换它，以保证集群的高可用

分布式锁

缓存穿透

缓存穿透是指用户不断请求缓存和数据库中都没有的信息（查询一个根本不存在的数据！），由于缓存是不命中时被动写的，并且出于容错考虑，如果从存储层查不到数据则不写入缓存，这将导致这个不存在的数据每次请求都要到存储层去查询，失去了缓存的意义。在流量大时，可能DB就挂掉了，要是有人利用不存在的key频繁攻击应用，如发起请求为id为“-1”的数据或id为特别大不存在的数据。就会导致数据库压力过大

解决方案：

  1. 接口层增加校验，如用户鉴权校验，id做基础校验，id<=0的直接拦截；

  2. 从缓存查不到，在数据库中也查不到的数据，这时也可以将key-value写为key-null（比如一个请求查询id=-1，查不到数据库后照样将该值缓存到redis中：redis中缓存一个key为-1，value为null的键值对），同时必须设置缓存时间（防止万一以后真的添加了对应id的数据行，如果不设置过期时间直接走缓存就会返回一个null值），缓存有效时间可以设置短点，如30秒（设置太长会导致正常情况也没法使用）。这样可以防止攻击用户反复用同一个id暴力攻击；

  3. [布隆过滤器][https://javaguide.cn/cs-basics/data-structure/bloom-filter.html]：bloomfilter就类似于一个hashset，用于快速判某个元素是否存在于集合中，其典型的应用场景就是快速判断一个key是否存在于某容器，不存在就直接返回。布隆过滤器的关键就在于hash算法和容器大小

比如当某一个字符串存储要加入到布隆过滤器中时，该字符串首先由多个哈希函数生成不同的哈希值，然后将对应的位数组的下标设置为 1（当位数组初始化时，所有位置均为 0）。当第二次存储相同字符串时，因为先前的对应位置已设置为 1，所以很容易知道此值已经存在（去重非常方便）；

如果我们需要判断某个字符串是否在布隆过滤器中时，只需要对给定字符串再次进行相同的哈希计算，得到值之后判断位数组中的每个元素是否都为 1，如果值都为 1，那么说明这个值在布隆过滤器中，如果存在一个值不为 1，说明该元素不在布隆过滤器中。

综上可以得出：布隆过滤器说某个元素存在，小概率会误判。布隆过滤器说某个元素不在，那么这个元素一定不存在！

缓存击穿

缓存击穿是指大量并发访问同一个热点数据（缓存雪崩是多个数据）时，当该热点数据缓存失效后同时请求数据库，瞬间耗尽数据库资源，导致数据库压力过大

解决方案：

  1. 设置热点数据永远不过期
  2. 仅对查询数据库的业务逻辑代码块来设置同步锁synchronized(this)

比如查询逻辑调用的service为PublishServiceImpl,由于在对应的ServiceImpl上设置了**@Service注解，故该类的实例会被注入到Spring容器中，而Spring默认Bean为单例的（Singleton**）,因此如果多个线程请求过来共享这一个实例，即共享这一个锁

缓存雪崩

缓存雪崩是指缓存中大批数据一起过期，此时如果外来请求过多，使得大量请求直接查询数据库，导致数据库压力过大甚至宕机。和缓存击穿不同的是，缓存击穿指并发查同一条数据，缓存雪崩是大量不同数据一起过期了，

解决方案：

  1. 对缓存数据的过期时间设置随机，防止同一时间大量数据过期现象发生。

  2. 如果缓存数据库是分布式部署，将热点数据均匀分布在不同的缓存数据库中。

  3. 使用synchronized同步锁控制查询数据库的线程（并发度很低）

  4. 不用等到请求到来再去查询数据库时才存入缓存，可以提前将数据存入缓存中，后台通常有定时任务将数据库中的数据存入缓存

Redisson分布式锁

（1）基于synchronized的同步锁问题：

synchronized锁的是当前JVM，如果多个JVM进程同时访问一个接口，此时再使用同步锁是锁不住的：即一个同步锁程序只能保证同一个虚拟机中多个线程只有一个线程去查数据库，如果高并发情况下通过网关负载均衡转发给多个虚拟机，此时就会存在多个线程去查询数据库的情况，因为同步锁（虚拟机内部的锁）无法跨虚拟机保证同步执行

（2）基于setnx实现分布式锁的问题：

SETNX 命令是 Redis 中用于实现分布式锁的经典方案之一。它可以将一个指定的 key 设置为一个特定的值，只有当这个 key 不存在时才能设置成功，但是当多个JVM同时尝试获取同一个锁时，其中一个JVM进程获取成功后需要设置锁的过期时间，过期时间设置过小会导致当前请求还未处理完锁就被释放；时间设置过长又会导致资源浪费；而如果采取处理完请求后手动释放锁的方法，又会产生原子性问题：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
if(缓存里有)
{
返回缓存里数据
}else{
获取分布式锁： set lock 01 NX
if(获取锁成功)
{
try{
查询数据库
}
finally{
# if(redis.call("get","lock")=="01"){
# 释放锁: redis.call("del","lock")
                }
           }
      }
}

可以看出redis在执行带#的两行代码时不能保证原子性，故需要通过lua脚本保证原子性：

解锁脚本的一个例子将类似于以下：

1
2
3
4
5
6
if redis.call("get",KEYS[1]) == ARGV[1]
then
    return redis.call("del",KEYS[1])
else
    return 0
end

Redisson的加锁机制如下图所示，线程去获取锁，获取成功则执行lua脚本，保存数据到redis数据库。如果获取失败: 一直通过while循环尝试获取锁(可自定义等待时间，超时后返回失败)，获取成功后，执行lua脚本，保存数据到redis：

如果拿到分布式锁的节点宕机，且这个锁正好处于锁住的状态时，会出现锁死的状态，为了避免这种情况的发生，锁都会设置一个过期时间。这样也存在一个问题，加入一个线程拿到了锁设置了30s超时，在30s后这个线程还没有执行完毕，锁超时释放了，就会导致问题，Redisson给出了自己的答案，自动延期(Watch Dog)机制:

Redisson提供了一个监控锁的看门狗线程，它的作用是在Redisson实例被关闭前，不断的延长锁的有效期，也就是说，如果一个拿到锁的线程一直没有完成逻辑，那么看门狗会帮助线程不断的延长锁超时时间，锁不会因为超时而被释放。默认情况下，看门狗的续期时间是30s，也可以通过修改Config.lockWatchdogTimeout来另行指定

项目中使用Redisson实现分布式锁核心代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
 public CoursePublish getCoursePublishCache(Long courseId){
            //先从缓存中查询
            String jsonString = (String) redisTemplate.opsForValue().get("course:" + courseId);
            if(StringUtils.isNotEmpty(jsonString)){
             System.out.println("========从缓存中查询===========");
                //将json转成对象返回
                CoursePublish coursePublish = JSON.parseObject(jsonString, CoursePublish.class);
                return coursePublish;
            }else{
                //使用setnx向redis设置一个key，谁设置成功谁拿到了锁
//             Boolean lock001 = redisTemplate.opsForValue().setIfAbsent("lock001", "001",300,TimeUnit.SECONDS);
                //使用redisson获取锁
                RLock lock = redissonClient.getLock("coursequery:" + courseId);
                //获取分布式锁
                lock.lock();
                //拿到锁的才会执行try finally代码块
                try{
                        //Thread.sleep(35000);
                        //再次从缓存中查询一下
                        jsonString = (String) redisTemplate.opsForValue().get("course:" + courseId);
                        if(StringUtils.isNotEmpty(jsonString)){
                            //将json转成对象返回
                            CoursePublish coursePublish = JSON.parseObject(jsonString, CoursePublish.class);
                            return coursePublish;
                        }
                        System.out.println("从数据库查询...");
                        //如果缓存中没有，要从数据库查询
                        CoursePublish coursePublish = coursePublishMapper.selectById(courseId);
                        //将从数据库查询到的数据存入缓存 ,缓存中设置Key为 "course:id"
                        redisTemplate.opsForValue().set("course:" + courseId,JSON.toJSONString(coursePublish),300, TimeUnit.SECONDS);
                        return coursePublish ;
                }finally {
                    //释放锁
                    lock.unlock();  //Redisson封装了底层执行lua脚本来实现解锁原子性的过程
                }
            }
    }