集群

这是《Redis设计与实现》系列的文章，系列导航：Redis设计与实现笔记

创建集群

通过 CLUSTER NODE 命令可以查看当前集群中的节点。刚启动时，默认每一台节点都是一个集群。

sequenceDiagram
participant r0 as redis.cn 7000
participant r1 as redis.cn 7001
participant r2 as redis.cn 7002

note over r1: CLUSTER MEET redis.cn 7000
r1 ->>+ r0: 握手
r0 ->>- r1: 响应握手

note over r1: CLUSTER MEET redis.cn 7002
r1 ->>+ r2: 握手
r2 ->>- r1: 响应握手

如上图所示，登录 7001，然后输出相应的指令，请求和7000、7002搭建一个集群。

graph TD;

subgraph 集群
7001
7002
7003
end

集群的数据结构

节点会继续使用所有在单机模式中使用的服务器组件，而那些只有在集群模式下才能用到的数据，节点将他们保存到了 cluster.h/clusterNode 、cluster.h/clusterLink 、cluster.h/clusterState结构中，如下图所示。

前面两个结构主要记录的是其他节点的信息，而clusterState则是集群的信息。

这里有两个纪元：

这两个纪元分别用在哪里，有什么不同？

MEET命令的实现

上面的那张 MEET 的时序图非常的概括，比如：

1. 怎么握手，怎么响应握手
2. 接收方怎么开始与其他节点进行 MEET

这里在该图的基础上补充一些细节

sequenceDiagram
participant r0 as redis.cn 7000
participant r1 as redis.cn 7001

note over r1: CLUSTER MEET redis.cn 7000

r1 ->> r1: 创建7000的clusterNode结构

r1 ->>+ r0: MEET

r0 ->> r0: 创建7001的clusterNode结构

r0 ->>- r1: PONG

note over r1: 我知道了你已经收到了

r1 ->> r0: PING

note over r0: 我知道你已经收到了

这里互相确认有一点三次握手的感觉

节点数据库的实现

集群节点保存键值对以及过期时间的方式与单机 Redis 服务器的方式完全相同。

但是一个区别是，节点只能使用0号数据库。

利用集群可以实现分区的功能，从而减少单台服务器的业务量。那么集群的首要任务就是如何保证一致性。Redis 采用了槽指派的模式进行分区，类似于一致性哈希的做法。

Redis 集群将整个数据库分为 16384 个槽，数据库中每一个键都属于者16384个槽中的一个，集群中的每个节点可以处理0到16384个槽。

操作

只有所有槽都有节点在处理时，集群才处于上线状态，否则，处于下线状态。

可以用 CLUSTER INFO 查看集群状态（我们之前的集群就没有分配槽，所以是下线状态）。

可以用 CLUSTER ADDSLOTS xxx 命令进行槽的分配。

数据结构

clusterNode结构中：

• numslots：表示一个节点负责的槽数量
• slots：位数组，每一位记录了这个节点归不归我管，比如下面的表格，表示这个节点只负责 1 ~ 8 号槽

clusterState结构中：

clusterNode* slots[16384] 结构中记录了所有槽的指派信息：

• NULL 表示未被指派
• 否则指向该节点的结构

ADDSLOTS命令实现

先遍历一遍，看有没有已经分配过了的，如果有则直接失败。

否则，设置更新上述的两个结构。

伪码便于理解：

槽指派信息传播

一个节点除了会记录自己的上述信息外，还会将这个数组通过消息发送给集群中的其他节点。

其他节点收到后，进行保存或更新。

执行流程

集群模式的一个重要的不同是，数据被分配在了不同的节点上，所以接收到请求的服务器未必能对该请求进行处理，因此多了一个寻找有能力处理的服务器的过程：

sequenceDiagram
participant a as NodeA
participant c as Client
participant b as NodeB
c ->>+ a: GET name
a ->> a: 计算 CRC16(name) % 16384
a ->> a: clusterNode.slots 中该值是否是我负责
alt slots[i] = 0 是我的工作
a ->> c : name="张三"
else slots[i] = 0 不是我的工作
a ->> c : MOVE NodeB
c ->> b : GET name
note over b: 省略若干判断操作
b ->> c : name="张三"
end

MOVE指令

指令格式为：

MOVE <slot> <ip>:<port>

和 HTTP 请求的重定向有些类似

数据结构

保存槽的分配情况：

clusterState.slots_to_keys 是一个跳表，用来保存槽和键之间的关系。跳表的分值是一个槽号，而节点的成员都是数据库键。每当节点往数据库中插入一个新的键值对时，节点就会将这个键以及键的槽号关联到这个跳表中。

这么做的目的是，方便我们找到某一个槽值对应的键，例如命令 CLUSTER GETKEYSINSLOT。（话说如果要重新分配槽的话，不就有这个需求了！）

执行流程

由Redis的集群管理软件 redis-trib 负责执行，流程如下：

sequenceDiagram
autonumber
loop 对于迁移的每个槽
participant s as source node
participant rt as redis-trib
participant t as target node
note over rt: 先通知你们俩都作好准备啊
rt ->> t: 你从获得号槽
rt ->> s: 你把号槽给
note over rt: 我要开始迁移了！
loop 只要还有值没有迁移
rt ->> s: 返回最多个槽的键
rt ->> t: 这是键的信息，你保存一下
end
note over rt: 迁移完了，通知一下
end

图注：

最后的通知步骤，书上说是：向集群中任意一个节点发送 CLUSTER SETSLOT <slot> NODE <target_id> 命令。我理解之为 Gossip 协议的方式。

IMPORTING 命令的实现

clusterState 结构的 importing_slots_from 数组记录了当前节点正在从其他节点导入的槽。如果 importing_slots_from[i] 的值不为 NULL，而是指向一个 clusterNode结构，则表明当前节点正在从 clusterNode 所代表的节点导入槽i。

MIGRATEING 命令的实现

类似地，clusterState 结构的 migrating_slots_to 数组记录了当前节点正在迁移至其他节点的槽。其情况和上述命令一样。

ASK错误

在迁移的过程中，难免会出现一种情况：某个槽值的键只迁移了一部分，有一部分还保存在原来的节点，而另一部分已经保存在目标节点了。

为了处理这种情况，我们需要一些机制来进行处理：

sequenceDiagram
participant c as Client
participant s1 as Node1
participant s2 as Node2
note over s1,s2: Node1正在给Node2迁移槽
c ->>+ s1: GET name
s1 ->> s1: 计算对应槽值，发现该槽是我负责的
s1 ->> s1: 查找槽值，发现没找到，发现这个槽正在转移
s1 ->>- c: ASK Node2
c ->>+ c: 打开我的ASING标识
c ->>+ s2: ASKING
c ->> s2: GET name
s2 ->> s2: 这个槽不归我管，给它回个MOVE吧
s2 ->> s2: 不对，你是ASKING，我再找找
s2 ->>- c: 找到了, name="张三"
c ->>- c: 好滴，关闭我的ASKING标识

• 收到 ASK 后，会打开自己的 ASKING 标识
• 在发送请求前先发送一次 ASKING
• ASK 使用后就会关闭

可以给集群中的节点设置从节点，从而提高系统的容错性和高可用。

节点复制的方法

1. 节点收到 CLUSTER REPLICATE <node_id>，开始进行复制
2. 修改 clusterState.myself.slaveof 指针，指向主节点
3. 修改 clusterState.myself.flag ，关闭 REDIS_NODE_MASTER 标识，打开 REDIS_NODE_SLAVE 标识
4. 进行复制，相当于执行 SLAVEOF <master_ip> <master_port>
5. 将这一情况通过消息发送给集群中的所有节点

故障检测

这里的故障检测和 Sentinel 的故障检测是很相似的，如下：

1. 集群的节点互相发送 PING，并接收 PONG
2. 如果一定时间没有收到恢复，就标记为疑似下线状态(PFAIL)
3. 集群中的节点会相互发消息交换状态信息
4. 如果一个集群里半数以上负责处理槽的主节点都标记为下线，则其被标记为下线(FAIL)
5. 广播下线消息

故障转移

当一个从节点发现自己的老大挂了，就要选举一个新的老大，并给老大安排后事。

1. 选举一个节点
2. 这个新的节点将执行 SLAVEOF no one，成为新的老大
3. 新的老大会把原来老大的槽全部指派给自己
4. 新的老大会向别人（集群）广播一条 PONG 消息，宣告自己的地位
5. 新的老大开始接收和处理命令请求，故障转移完成

那么谁来当新的老大呢，如何选举？基于Raft算法的领头选举

和之前 Sentinel 部分的处理情况非常的类似，这里我就不再次描述了，贴上官方文档，供大家学习：Redis cluster specification | Redis

消息介绍

有五种消息：

• MEET：接收到客户端发送的 CLUSTER MEET 指令时发送，请求接收者加入当前的集群中
• PING：集群中的每个节点默认每隔一秒就会从已知节点列表中随机选出5个节点，然后对这五个节点中最长时间没有发过 PING 消息的节点发送 PING 消息，以此检测被选中的节点是否在线。除此之外，对超过 cluster-node-timeout 时间没有发送过节点的也会发送。
• PONG：一是对 MEET 或 PING 命令的响应；二是通过向集群广播 PONG 来刷新其他节点对自己的认识，例如故障转移后的主节点。
• FAIL：当一个主节点 A 判断另一个节点 B 已经进入 FAIL 状态时，会广播一条关于 B 的 FAIL 消息，所有收到这条消息的节点会将 B 标记为下线（前文提到过）
• PUBLISH：当一个节点接收到一个 PUBLISH 命令时，节点会执行这个命令，并向集群广播一条 PUBLISH 命令，所有接收到这条信息的节点会同样进行这个过程。

一条消息由消息头和消息正文组成。

消息头

每个消息头由一个 cluster.h/clusterMsg 结构表示：

typedef struct {
    char sig[4];        /* Siganture "RCmb" (Redis Cluster message bus). */
    // 消息的长度（包括这个消息头的长度和消息正文的长度）
    uint32_t totlen;    /* Total length of this message */
    uint16_t ver;       /* Protocol version, currently set to 0. */
    uint16_t notused0;  /* 2 bytes not used. */
    // 消息的类型
    uint16_t type;
    // 消息正文包含的节点信息数量
    // 只在发送 MEET 、 PING 和 PONG 这三种 Gossip 协议消息时使用
    uint16_t count;
    // 消息发送者的配置纪元
    uint64_t currentEpoch;

    // 如果消息发送者是一个主节点，那么这里记录的是消息发送者的配置纪元
    // 如果消息发送者是一个从节点，那么这里记录的是消息发送者正在复制的主节点的配置纪元
    uint64_t configEpoch;

    // 节点的复制偏移量
    uint64_t offset;
    // 消息发送者的名字（ID）
    char sender[REDIS_CLUSTER_NAMELEN];

    // 消息发送者目前的槽指派信息
    unsigned char myslots[REDIS_CLUSTER_SLOTS/8];

    // 如果消息发送者是一个从节点，那么这里记录的是消息发送者正在复制的主节点的名字
    // 如果消息发送者是一个主节点，那么这里记录的是 REDIS_NODE_NULL_NAME
    // （一个 40 字节长，值全为 0 的字节数组）
    char slaveof[REDIS_CLUSTER_NAMELEN];

    char notused1[32];
    // 消息发送者的端口号
    uint16_t port;
    // 消息发送者的标识值
    uint16_t flags;
    // 消息发送者所处集群的状态
    unsigned char state;
    // 消息标志
    unsigned char mflags[3];
    // 消息的正文（或者说，内容）
    union clusterMsgData data;

} clusterMsg;

这些属性记录了发送者自身的节点信息，接收者会根据这些信息，在 clusterState.nodes 字典中找到发送者对应的 clusterNode 结构，并对结构进行更新。

消息体

上文的最后一个属性 union clusterMsgData data 指向联合结构，这个结构就是消息的正文：

MEET、PING、PONG

Redis 集群中各个节点通过 Gossip 协议来交换各自关于不同节点的状态信息，其中 Gossip 协议由 MEET、PING、PONG三种消息实现，这三种消息的正文都由两个 cluster.h/clusterMsgDataGossip 结构组成：

注意到MEET、PING、PONG 都使用相同的消息正文，所以节点通过消息头的 type 属性来判断一条消息是 MEET 消息、PING 消息还是 PONG 消息。

每次发送 MEET、PING、PONG 三种消息时，发送者都从自己的已知节点列表中随机选择出两个节点（可以是主或从），并将这两个被选中的节点信息分别保存到两个 clusterMsgDataGossip 结构中：

typedef struct {

    // 节点的名字
    // 在刚开始的时候，节点的名字会是随机的
    // 当 MEET 信息发送并得到回复之后，集群就会为节点设置正式的名字
    char nodename[REDIS_CLUSTER_NAMELEN];

    // 最后一次向该节点发送 PING 消息的时间戳
    uint32_t ping_sent;

    // 最后一次从该节点接收到 PONG 消息的时间戳
    uint32_t pong_received;

    // 节点的 IP 地址
    char ip[REDIS_IP_STR_LEN];    /* IP address last time it was seen */

    // 节点的端口号
    uint16_t port;  /* port last time it was seen */

    // 节点的标识值
    uint16_t flags;

    // 对齐字节，不使用
    uint32_t notused; /* for 64 bit alignment */

} clusterMsgDataGossip;

过程分为认识和不认识：

sequenceDiagram
participant s as A
participant r as B
participant r3 as D

s ->> s: RandomGetTwoNode
s ->> r: PING(C, D)
r ->> r: 我认识C, 所以我更新他的信息
r ->> r3: 不认识D，进行握手
r ->> r: RandomGetTwoNode
r ->> s: PONG(E, F)

有一个疑惑：源码中 clusterMsgDataGossip 大小明明为 1，怎么保存两个节点的信息的。

备注：前面提到的 PING 每秒选五个节点进行发送，这里提到的是每次发送这三种信息时附带随机的两个节点的信息。

FAIL

FAIL 消息用来宣告某一个节点的失效，由于这个消息属于“八百里加急”，需要让所有节点立即知道。而当节点数量比较大的时候延迟较大，所以不适合使用 Gossip 协议。

cluster.h/clusterMsgDataFail 的结构比较简单，仅用名称标识进行唯一标识：

typedef struct {

    // 下线节点的名字
    char nodename[REDIS_CLUSTER_NAMELEN];

} clusterMsgDataFail;

PUBLISH

当客户端向集群中某个节点发送命令：

PUBLISH <channel> <message>

接收到 PUBLIHS 命令的节点不仅会向 channel 频道发送消息 message，还会向集群广播一条 PUBLISH 消息。而其他接收到消息的节点的也都会向 channel 频道发送 message 消息。

原书：为什么不直接向节点广播 PUBLISH 命令？

要让集群所有节点都执行相同的 PUBLISH 命令，最简单的方法就是向所有节点广播相同的 PUBLISH 命令，这也是 Redis 在复制 PUBLISH 命令时使用的方式，不过这种做法并不符合 Redis 集群的 “各个节点通过发送和接收消息来进行通信”这一规则，所以节点没有采取广播 PUBLISH 命令的方法。

消息的结构：

typedef struct {

    // 频道名长度
    uint32_t channel_len;

    // 消息长度
    uint32_t message_len;

    // 消息内容，格式为 频道名+消息
    // bulk_data[0:channel_len-1] 为频道名
    // bulk_data[channel_len:channel_len+message_len-1] 为消息
    unsigned char bulk_data[8]; /* defined as 8 just for alignment concerns. */

} clusterMsgDataPublish;