redis源码分析--cluster消息

Posted 2023-01-04 yang-zd

Redis集群消息

  作为支持集群模式的缓存系统，Redis集群中的各个节点需要定期地进行通信，以维持各个节点关于其它节点信息的实时性与一致性。如前一篇文章介绍的，Redis在专用的端口监听集群其它节点的连接，将集群内部的的通信与客户端的通信区分开来，任意两个节点之间建立了两个tcp连接，形成一条全双工的通道。这篇文章将从集群消息方面进行介绍，主要介绍消息的格式、种类与不同场景下的消息处理。

1. 消息格式

  首先，Redis集群通信使用的消息可分为消息头与消息体两部分：消息头包含了发送消息的节点的具体信息，每一个消息必须拥有完整的消息头；消息体根据消息类型的不同具有不同的内容，也有一些类型的消息不包含消息体。完整的消息格式定义clusterMsg如下所示(cluster.h中)：

typedef struct 
    char sig[4];        /* Signature "RCmb" (Redis Cluster message bus). */
    uint32_t totlen;    /* Total length of this message */
    uint16_t ver;       /* Protocol version, currently set to 1. */
    uint16_t port;      /* TCP base port number. */
    uint16_t type;      /* Message type */
    uint16_t count;     /* Only used for some kind of messages. */
    uint64_t currentEpoch;  /* The epoch accordingly to the sending node. */
    uint64_t configEpoch;   /* The config epoch if it‘s a master, or the last
                               epoch advertised by its master if it is a
                               slave. */
    uint64_t offset;    /* Master replication offset if node is a master or
                           processed replication offset if node is a slave. */
    char sender[CLUSTER_NAMELEN]; /* Name of the sender node */
    unsigned char myslots[CLUSTER_SLOTS/8];
    char slaveof[CLUSTER_NAMELEN];
    char myip[NET_IP_STR_LEN];    /* Sender IP, if not all zeroed. */
    char notused1[34];  /* 34 bytes reserved for future usage. */
    uint16_t cport;      /* Sender TCP cluster bus port */
    uint16_t flags;      /* Sender node flags */
    unsigned char state; /* Cluster state from the POV of the sender */
    unsigned char mflags[3]; /* Message flags: CLUSTERMSG_FLAG[012]_... */
    union clusterMsgData data;
 clusterMsg;

最后的data成员即消息体，它是一个union结构，根据消息种类使用不同的消息体。

totlen是消息的总长度，接收方在读取了消息的前8bytes后即可知道消息的总长度。

ip,port,cport即发送消息的节点的ip，数据端口与集群端口

type代表了消息的类型，由宏定义CLUSTERMSG_TYPE_*定义

sender是发送消息的节点的nameid，用于查找clusterNode结构

myslots是发送节点(发送节点是master)或者发送节点的master节点对应的clustreNode结构中的slots，用于同步各个节点的slots信息

slaveof，如果发送消息的节点是slave，那么slaveof存储它的master的nameid

flags代表发送节点的状态，如该节点是slave还是master的标志。

  消息体data根据type的取值具有不同的结构，Redis中定义了如下的消息类型：

/* Message types.
 *
 * Note that the PING, PONG and MEET messages are actually the same exact
 * kind of packet. PONG is the reply to ping, in the exact format as a PING,
 * while MEET is a special PING that forces the receiver to add the sender
 * as a node (if it is not already in the list). */
#define CLUSTERMSG_TYPE_PING 0          /* Ping */
#define CLUSTERMSG_TYPE_PONG 1          /* Pong (reply to Ping) */
#define CLUSTERMSG_TYPE_MEET 2          /* Meet "let‘s join" message */
#define CLUSTERMSG_TYPE_FAIL 3          /* Mark node xxx as failing */
#define CLUSTERMSG_TYPE_PUBLISH 4       /* Pub/Sub Publish propagation */
#define CLUSTERMSG_TYPE_FAILOVER_AUTH_REQUEST 5 /* May I failover? */
#define CLUSTERMSG_TYPE_FAILOVER_AUTH_ACK 6     /* Yes, you have my vote */
#define CLUSTERMSG_TYPE_UPDATE 7        /* Another node slots configuration */
#define CLUSTERMSG_TYPE_MFSTART 8       /* Pause clients for manual failover */
#define CLUSTERMSG_TYPE_MODULE 9        /* Module cluster API message. */
#define CLUSTERMSG_TYPE_COUNT 10        /* Total number of message types. */

  1）CLUSTERMSG_TYPE_PING/ClUSTERMSG_TYPE_PONG/ClUSTERMSG_TYPE_MEET的消息体是如下结构的一个数组，数组长度由消息头中的count字段表示。

typedef struct 
    char nodename[CLUSTER_NAMELEN];
    uint32_t ping_sent;
    uint32_t pong_received;
    char ip[NET_IP_STR_LEN];  /* IP address last time it was seen */
    uint16_t port;              /* base port last time it was seen */
    uint16_t cport;             /* cluster port last time it was seen */
    uint16_t flags;             /* node->flags copy */
    uint32_t notused1;
 clusterMsgDataGossip;

每一个clusterMsgDataGossip实例都表示了一个发送节点知道的节点的基本信息，包含nameid、ip、port、cport，flag表示该节点的状态，如是否挂掉等，而ping_sent与pong_received代表了发送节点与该节点的通信状态。ping/pong类消息是集群内正常通信使用最多的消息类型。

2) CLUSTERMSG_TYPE_FAIL的消息体定义如下：

typedef struct 
    char nodename[CLUSTER_NAMELEN];
 clusterMsgDataFail;

它在master节点判定一个节点离线时发送，消息体仅包含离线节点的nameid。

3) CLUSTERMSG_TYPE_UPDATE的定义如下：

typedef struct 
    uint64_t configEpoch; /* Config epoch of the specified instance. */
    char nodename[CLUSTER_NAMELEN]; /* Name of the slots owner. */
    unsigned char slots[CLUSTER_SLOTS/8]; /* Slots bitmap. */
 clusterMsgDataUpdate;

它用于通知接收节点去更新消息体中nodename指定的节点负责的slots。

4) CLUSTERMSG_TYPE_FAILOVER_AUTH_REQUEST/CLUSTERMSG_TYPE_FAILOVER_AUTH_ACK及CLUSTERMSG_TYPE_MFSTART类消息在slave需要接替master节点作为新的主节点时使用，没有消息体。

  以上几类是集群中主要使用的消息类型，后面将介绍它们的应用场景，其它消息类型与集群状态无关，暂不作介绍。

2. 发现新节点

  当集群需要扩容时，新启动的节点并不会自动加入到原有集群中，通常需要客户端执行命令meet，并指向需要连接的节点的ip、port、cport参数，meet命令执行流程如下：

1. 执行meet命令的节点会新建一个clusterNode实例，对应参数指定的ip、port、cport，此时clusterNode实例的flag标记为CLUSTER_NODE_HANDSHAKE及CLUSTER_NODE_MEET状态。
2.  执行命令的节点在serverCron中为该clusterNode新建link，连接到指定节点，并向其发送ClUSTERMSG_TYPE_MEET消息，去除clusterNode的CLUSTER_NODE_MEET标记。
3. 接收到ClUSTERMSG_TYPE_MEET消息的节点会为发送节点建立新的clusterNode实例，此时该clusterNode处于CLUSTER_NODE_HANDSHAKE状态。
4.  接收方向发送方回应ClUSTERMSG_TYPE_PONG消息
5.  发送方收到ClUSTERMSG_TYPE_PONG消息，去除对应clusterNode的CLUSTER_NODE_HANDSHAKE标记，并将clusterNode的nameid更新为pong消息头的nameid。
6. 接收方在serverCon中主动与发送节点建立连接，将link保存到它新建立的clusterNode实例中，并发送ClUSTERMSG_TYPE_PING消息。
7. 发送方收到ClUSTERMSG_TYPE_PING消息，回应ClUSTERMSG_TYPE_PONG消息
8. 接收方收到ClUSTERMSG_TYPE_PONG消息后，去除对应clusterNode的CLUSTER_NODE_HANDSHAKE标记，并将clusterNode的nameid更新为pong消息头的nameid。

通过以上步骤，两个互不认识的节点建立了一个全双工的通道。

3. 发现节点离线

  serverCron中会定期地向nodes字典中的clusterNode对应节点发送ping消息，如果超时未收到pong消息作为回应，那么将clusterNode中的flag标记为CLUSTER_NODE_PFAIL，然后该状态会在ping/pong/meet消息的消息体中扩散到其它节点，开启一个节点离线的判断流程，具体如下：

1. nodeA向nodeB发送的ping消息超时未收到响应，将nodeB在nodeA中对应的clusterNode实例的flag标记为CLUSTER_NODE_PFAIL。
2. nodeA向其它仍然在线的节点发送ping/pong/meet消息，将nodeB的状态加入到消息体之中，以一个clusterMsgDataGossip实例表示。
3. 其它节点在收到nodeA的消息后，调用clusterProcessGossipSection处理消息体。
4. 其它节点查找本节点中clusterMsgDataGossip对应的clusterNode实例，并将该结点离线的事件记录到fail_reports链表上。 (这里假设其它节点认识nodeA，并且nodeA是master，否则CLUSTER_NODE_PFAIL的状态不会被接受)。
5. 其它节点统计clusterNode实例的fail_reports链表上报告该节点离线的master数目，如果数目超过了所有master节点的一半，将其标记为CLUSTER_NODE_FAIL状态。此外，如果当前做出如此判断的节点是master节点，那么向其它节点发送CLUSTERMSG_TYPE_FAIL消息，消息体中带有离线节点的nameid。
6. 其它节点收到CLUSTERMSG_TYPE_FAIL消息后，立刻查找消息体中nameid对应的clusterNode，将其标记为CLUSTERMSG_TYPE_FAIL状态。

  clusterCron中定期会执行clusterHandleSlaveFailover函数，这个函数中如果有slave发现了它的master离线，那么便会开启一段替换主节点的流程。

  离线的节点的clusterNode并不会被删除，而是它们的link结构会被释放，然后serverCron中会不断尝试重新建立连接，一旦重新建立的连接收到pong回应，那么该clusterNode实例会清除CLUSTERMSG_TYPE_FAIL或者CLUSTERMSG_TYPE_PFAIL标记，然后新的状态会跟随消息进行扩散，其它节点收到消息后删除对应clusterNode中的fail_reports链表上该节点的报告。

4. ping/pong/meet消息的消息体

这3类消息的消息体clusterMsgDataGossip除了报告离线节点的作用外，还有以下作用：

1. 如果接收消息的节点查找不到clusterMsgDataGossip实例中的nameid，那么开启一段发现新节点的流程。
2. 如果接收消息的节点查找到clusterMsgDataGossip对应的clusterNode处于离线状态，但是clusterMsgDataGossip中状态却是在线，并且两个结构中记录的ip、port、cport不同，那么更新本地记录的ip、port、cport，以便尝试重新连接。

5. 更新slots配置

  每一个集群消息都在消息头携带了发送节点的信息，其中slots是发送节点或者其master节点负责的slot的bit掩码，接收方收到收到消息后将查找本地对应发送节点的clusterNode结构，对比两份slots掩码是否有所不同。消息头与clusterNode中都有一个参数configEpoch记录信息的更新时间，值越大表示配置越新，两个节点中slots不相同，以新的配置为准。根据情况做如下处理：

1. 发送方是主节点，并且消息头中的slots与本地对应clusterNode中的slots不相同，调用函数clusterUpdateSlotsConfigWith更新clusterState与clusterNode中的slots。
2. 如果消息头中的slots声明的slot在接收节点中发现由其它节点clusterNode_x负责，并且clusterNode_x中记录的configEpoch值更大，那么向发送方回应CLUSTERMSG_TYPE_UPDATE消息，通知它更新它的slots配置。
3. 接收到CLUSTERMSG_TYPE_UPDATE消息的节点调用clusterUpdateSlotsConfigWith函数更新它的slots配置。

6. 替换主节点

在serverCron中定期执行函数clusterHandleSlaveFailover，检查是否需要以本节点替换它的master，当集群中的master离线，或者客户端执行命令主动替换master节点，集群开启替换流程：

1. 执行替换的节点广播发送CLUSTERMSG_TYPE_FAILOVER_AUTH_REQUEST类型的消息，请求其它master节点回应它的替换请求。
2. 接收到CLUSTERMSG_TYPE_FAILOVER_AUTH_REQUEST消息的节点调用函数clusterSendFailoverAuthIfNeeded判断它是否回应请求。仅有效的master节点，并且判断发送方符合条件才回应CLUSTERMSG_TYPE_FAILOVER_AUTH_ACK消息。
3. 收到CLUSTERMSG_TYPE_FAILOVER_AUTH_ACK消息的节点增加它的授权记录，在clusterHandleSlaveFailover函数中判断授权数是否大于所有有效master节点数的一半，是则调用函数clusterFailoverReplaceYourMaster执行替换，否则继续等待。

一个master节点可能有多个slave，每一个slave将根据它数据的完整性决定它发送CLUSTERMSG_TYPE_FAILOVER_AUTH_REQUEST请求前需要等待的时间，数据完整的slave更早发送请求，其它master节点回应了某个slave的请求这个事件会被记录，它将不再回应针对同一个master的替换请求。