一万字详解 Redis Cluster Gossip 协议

Redis Cluster 是在 3.0 版本引入集群功能。为了让让集群中的每个实例都知道其他所有实例的状态信息，Redis 集群规定各个实例之间按照 Gossip 协议来通信传递信息。

上图展示了主从架构的 Redis Cluster 示意图，其中实线表示节点间的主从复制关系，而虚线表示各个节点之间的 Gossip 通信。

Redis Cluster 中的每个节点都维护一份自己视角下的当前整个集群的状态，主要包括：

1. 当前集群状态
2. 集群中各节点所负责的 slots信息，及其migrate状态
3. 集群中各节点的master-slave状态
4. 集群中各节点的存活状态及怀疑Fail状态

也就是说上面的信息，就是集群中Node相互八卦传播流言蜚语的内容主题，而且比较全面，既有自己的更有别人的，这么一来大家都相互传，最终信息就全面而且一致了。

Redis Cluster 的节点之间会相互发送多种消息，较为重要的如下所示：

• MEET：通过「cluster meet ip port」命令，已有集群的节点会向新的节点发送邀请，加入现有集群，然后新节点就会开始与其他节点进行通信；
• PING：节点按照配置的时间间隔向集群中其他节点发送 ping 消息，消息中带有自己的状态，还有自己维护的集群元数据，和部分其他节点的元数据；
• PONG: 节点用于回应 PING 和 MEET 的消息，结构和 PING 消息类似，也包含自己的状态和其他信息，也可以用于信息广播和更新；
• FAIL: 节点 PING 不通某节点后，会向集群所有节点广播该节点挂掉的消息。其他节点收到消息后标记已下线。

Redis 的源码中 cluster.h 文件定义了全部的消息类型，代码为 redis 4.0版本。

// 注意，PING 、 PONG 和 MEET 实际上是同一种消息。
// PONG 是对 PING 的回复，它的实际格式也为 PING 消息，
// 而 MEET 则是一种特殊的 PING 消息，用于强制消息的接收者将消息的发送者添加到集群中（如果节点尚未在节点列表中的话）
#define CLUSTERMSG_TYPE_PING 0          /* Ping 消息 */
#define CLUSTERMSG_TYPE_PONG 1          /* Pong 用于回复Ping */
#define CLUSTERMSG_TYPE_MEET 2          /* Meet 请求将某个节点添加到集群中 */
#define CLUSTERMSG_TYPE_FAIL 3          /* Fail 将某个节点标记为 FAIL */
#define CLUSTERMSG_TYPE_PUBLISH 4       /* 通过发布与订阅功能广播消息 */
#define CLUSTERMSG_TYPE_FAILOVER_AUTH_REQUEST 5 /* 请求进行故障转移操作，要求消息的接收者通过投票来支持消息的发送者 */
#define CLUSTERMSG_TYPE_FAILOVER_AUTH_ACK 6     /* 消息的接收者同意向消息的发送者投票 */
#define CLUSTERMSG_TYPE_UPDATE 7        /* slots 已经发生变化，消息发送者要求消息接收者进行相应的更新 */
#define CLUSTERMSG_TYPE_MFSTART 8       /* 为了进行手动故障转移，暂停各个客户端 */
#define CLUSTERMSG_TYPE_COUNT 9         /* 消息总数 */

通过上述这些消息，集群中的每一个实例都能获得其它所有实例的状态信息。这样一来，即使有新节点加入、节点故障、Slot 变更等事件发生，实例间也可以通过 PING、PONG 消息的传递，完成集群状态在每个实例上的同步。下面，我们依次来看看几种常见的场景。

首先，我们先来讲解一下其中涉及的数据结构，也就是上文提到的 ClusterNode 等结构。

每个节点都会维护一个 clusterState 结构，表示当前集群的整体状态，它的定义如下所示。

typedef struct clusterState {
   clusterNode *myself;  /* 当前节点的clusterNode信息 */
   ....
   dict *nodes;          /* name到clusterNode的字典 */
   ....
   clusterNode *slots[CLUSTER_SLOTS]; /* slot 和节点的对应关系*/
   ....
} clusterState;

它有三个比较关键的字段，具体示意图如下所示：

• myself 字段，是一个 clusterNode 结构，用来记录自己的状态；
• nodes 字典，记录一个 name 到 clusterNode 结构的映射，以此来记录其他节点的状态；
• slot 数组，记录slot 对应的节点 clusterNode结构。

clusterNode 结构保存了一个节点的当前状态，比如节点的创建时间、节点的名字、节点 当前的配置纪元、节点的IP地址和端口号等等。除此之外，clusterNode结构的 link 属性是一个clusterLink结构，该结构保存了连接节点所需的有关信息**，比如**套接字描述符，输入缓冲区和输出缓冲区。clusterNode 还有一个 fail_report 的列表，用来记录疑似下线报告。具体定义如下所示。

typedef struct clusterNode {
    mstime_t ctime; /* 创建节点的时间 */
    char name[CLUSTER_NAMELEN]; /* 节点的名字 */
    int flags;      /* 节点标识，标记节点角色或者状态，比如主节点从节点或者在线和下线 */
    uint64_t configEpoch; /* 当前节点已知的集群统一epoch */
    unsigned char slots[CLUSTER_SLOTS/8]; /* slots handled by this node */
    int numslots;   /* Number of slots handled by this node */
    int numslaves;  /* Number of slave nodes, if this is a master */
    struct clusterNode **slaves; /* pointers to slave nodes */
    struct clusterNode *slaveof; /* pointer to the master node. Note that it
                                    may be NULL even if the node is a slave
                                    if we don't have the master node in our
                                    tables. */
    mstime_t ping_sent;      /* 当前节点最后一次向该节点发送 PING 消息的时间 */
    mstime_t pong_received;  /* 当前节点最后一次收到该节点 PONG 消息的时间 */
    mstime_t fail_time;      /* FAIL 标志位被设置的时间 */
    mstime_t voted_time;     /* Last time we voted for a slave of this master */
    mstime_t repl_offset_time;  /* Unix time we received offset for this node */
    mstime_t orphaned_time;     /* Starting time of orphaned master condition */
    long long repl_offset;      /* 当前节点的repl便宜 */
    char ip[NET_IP_STR_LEN];  /* 节点的IP 地址 */
    int port;                   /* 端口 */
    int cport;                  /* 通信端口，一般是端口+1000 */
    clusterLink *link;          /* 和该节点的 tcp 连接 */
    list *fail_reports;         /* 下线记录列表 */
} clusterNode;

clusterNodeFailReport 是记录节点下线报告的结构体， node 是报告节点的信息，而 time 则代表着报告时间。

typedef struct clusterNodeFailReport {
    struct clusterNode *node;  /* 报告当前节点已经下线的节点 */
    mstime_t time;             /* 报告时间 */
} clusterNodeFailReport;

了解了 Reids 节点维护的数据结构体后，我们再来看节点进行通信的消息结构体。 通信消息最外侧的结构体为 clusterMsg，它包括了很多消息记录信息，包括 RCmb 标志位，消息总长度，消息协议版本，消息类型；它还包括了发送该消息节点的记录信息，比如节点名称，节点负责的slot信息，节点ip和端口等；最后它包含了一个 clusterMsgData 来携带具体类型的消息。

typedef struct {
    char sig[4];        /* 标志位，"RCmb" (Redis Cluster message bus). */
    uint32_t totlen;    /* 消息总长度 */
    uint16_t ver;       /* 消息协议版本 */
    uint16_t port;      /* 端口 */
    uint16_t type;      /* 消息类型 */
    uint16_t count;     /*  */
    uint64_t currentEpoch;  /* 表示本节点当前记录的整个集群的统一的epoch，用来决策选举投票等，与configEpoch不同的是：configEpoch表示的是master节点的唯一标志，currentEpoch是集群的唯一标志。 */
    uint64_t configEpoch;   /* 每个master节点都有一个唯一的configEpoch做标志，如果和其他master节点冲突，会强制自增使本节点在集群中唯一 */
    uint64_t offset;    /* 主从复制偏移相关信息，主节点和从节点含义不同 */
    char sender[CLUSTER_NAMELEN]; /* 发送节点的名称 */
    unsigned char myslots[CLUSTER_SLOTS/8]; /* 本节点负责的slots信息,16384/8个char数组，一共为16384bit */
    char slaveof[CLUSTER_NAMELEN]; /* master信息，假如本节点是slave节点的话，协议带有master信息 */
    char myip[NET_IP_STR_LEN];    /* IP */
    char notused1[34];  /* 保留字段 */
    uint16_t cport;      /* 集群的通信端口 */
    uint16_t flags;      /* 本节点当前的状态，比如 CLUSTER_NODE_HANDSHAKE、CLUSTER_NODE_MEET */
    unsigned char state; /* Cluster state from the POV of the sender */
    unsigned char mflags[3]; /* 本条消息的类型，目前只有两类：CLUSTERMSG_FLAG0_PAUSED、CLUSTERMSG_FLAG0_FORCEACK */
    union clusterMsgData data;
} clusterMsg;

clusterMsgData 是一个 union 结构体，它可以为 PING，MEET，PONG 或者 FAIL 等消息体。其中当消息为 PING、MEET 和 PONG 类型时，ping 字段是被赋值的，而是 FAIL 类型时，fail 字段是被赋值的。

// 注意这是 union 关键字
union clusterMsgData {
    /* PING, MEET 或者 PONG 消息时，ping 字段被赋值 */
    struct {
        /* Array of N clusterMsgDataGossip structures */
        clusterMsgDataGossip gossip[1];
    } ping;
    /*  FAIL 消息时，fail 被赋值 */
    struct {
        clusterMsgDataFail about;
    } fail;
    // .... 省略 publish 和 update 消息的字段
};

clusterMsgDataGossip 是 PING、PONG 和 MEET 消息的结构体，它会包括发送消息节点维护的其他节点信息，也就是上文中 clusterState 中 nodes 字段包含的信息，具体代码如下所示，你也会发现二者的字段是类似的。

typedef struct {
	/* 节点的名字，默认是随机的，MEET消息发送并得到回复后，集群会为该节点设置正式的名称*/
    char nodename[CLUSTER_NAMELEN]; 
    uint32_t ping_sent; /* 发送节点最后一次给接收节点发送 PING 消息的时间戳，收到对应 PONG 回复后会被赋值为0 */
    uint32_t pong_received; /* 发送节点最后一次收到接收节点发送 PONG 消息的时间戳 */
    char ip[NET_IP_STR_LEN];  /* IP address last time it was seen */
    uint16_t port;       /* IP*/       
    uint16_t cport;      /* 端口*/  
    uint16_t flags;      /* 标识*/ 
    uint32_t notused1;   /* 对齐字符*/
} clusterMsgDataGossip;

typedef struct {
    char nodename[CLUSTER_NAMELEN]; /* 下线节点的名字 */
} clusterMsgDataFail;

看完了节点维护的数据结构体和发送的消息结构体后，我们就来看看 Redis 的具体行为源码了。

以下是前方多次调用过的clusterSendPing()方法的源码，代码中有详细的注释，大家可以自行阅读。主要的操作就是将节点自身维护的 clusterState 转换为对应的消息结构体，。

/* 向指定节点发送一条 MEET 、 PING 或者 PONG 消息 */
void clusterSendPing(clusterLink *link, int type) {
    unsigned char *buf;
    clusterMsg *hdr;
    int gossipcount = 0; /* Number of gossip sections added so far. */
    int wanted; /* Number of gossip sections we want to append if possible. */
    int totlen; /* Total packet length. */
    // freshnodes 是用于发送 gossip 信息的计数器
    // 每次发送一条信息时，程序将 freshnodes 的值减一
    // 当 freshnodes 的数值小于等于 0 时，程序停止发送 gossip 信息
    // freshnodes 的数量是节点目前的 nodes 表中的节点数量减去 2 
    // 这里的 2 指两个节点，一个是 myself 节点（也即是发送信息的这个节点）
    // 另一个是接受 gossip 信息的节点
    int freshnodes = dictSize(server.cluster->nodes)-2;

    
    /* 计算要携带多少节点的信息，最少3个，最多 1/10 集群总节点数量*/
    wanted = floor(dictSize(server.cluster->nodes)/10);
    if (wanted < 3) wanted = 3;
    if (wanted > freshnodes) wanted = freshnodes;

    /* .... 省略 totlen 的计算等*/

    /* 如果发送的信息是 PING ，那么更新最后一次发送 PING 命令的时间戳 */
    if (link->node && type == CLUSTERMSG_TYPE_PING)
        link->node->ping_sent = mstime();
    /* 将当前节点的信息（比如名字、地址、端口号、负责处理的槽）记录到消息里面 */
    clusterBuildMessageHdr(hdr,type);

    /* Populate the gossip fields */
    int maxiterations = wanted*3;
    /* 每个节点有 freshnodes 次发送 gossip 信息的机会
       每次向目标节点发送 2 个被选中节点的 gossip 信息（gossipcount 计数） */
    while(freshnodes > 0 && gossipcount < wanted && maxiterations--) {
        /* 从 nodes 字典中随机选出一个节点（被选中节点） */
        dictEntry *de = dictGetRandomKey(server.cluster->nodes);
        clusterNode *this = dictGetVal(de);

        /* 以下节点不能作为被选中节点：
         * Myself:节点本身。
         * PFAIL状态的节点
         * 处于 HANDSHAKE 状态的节点。
         * 带有 NOADDR 标识的节点
         * 因为不处理任何 Slot 而被断开连接的节点 
         */
        if (this == myself) continue;
        if (this->flags & CLUSTER_NODE_PFAIL) continue;
        if (this->flags & (CLUSTER_NODE_HANDSHAKE|CLUSTER_NODE_NOADDR) ||
            (this->link == NULL && this->numslots == 0))
        {
            freshnodes--; /* Tecnically not correct, but saves CPU. */
            continue;
        }

        // 检查被选中节点是否已经在 hdr->data.ping.gossip 数组里面
        // 如果是的话说明这个节点之前已经被选中了
        // 不要再选中它（否则就会出现重复）
        if (clusterNodeIsInGossipSection(hdr,gossipcount,this)) continue;

        /* 这个被选中节点有效，计数器减一 */
        clusterSetGossipEntry(hdr,gossipcount,this);
        freshnodes--;
        gossipcount++;
    }

    /* .... 如果有 PFAIL 节点，最后添加 */


    /* 计算信息长度 */
    totlen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
    totlen += (sizeof(clusterMsgDataGossip)*gossipcount);
    /* 将被选中节点的数量（gossip 信息中包含了多少个节点的信息）记录在 count 属性里面*/
    hdr->count = htons(gossipcount);
    /* 将信息的长度记录到信息里面 */
    hdr->totlen = htonl(totlen);
    /* 发送网络请求 */
    clusterSendMessage(link,buf,totlen);
    zfree(buf);
}


void clusterSetGossipEntry(clusterMsg *hdr, int i, clusterNode *n) {
    clusterMsgDataGossip *gossip;
    /* 指向 gossip 信息结构 */
    gossip = &(hdr->data.ping.gossip[i]);
    /* 将被选中节点的名字记录到 gossip 信息 */   
    memcpy(gossip->nodename,n->name,CLUSTER_NAMELEN);
    /* 将被选中节点的 PING 命令发送时间戳记录到 gossip 信息 */
    gossip->ping_sent = htonl(n->ping_sent/1000);
    /* 将被选中节点的 PONG 命令回复的时间戳记录到 gossip 信息 */
    gossip->pong_received = htonl(n->pong_received/1000);
    /* 将被选中节点的 IP 记录到 gossip 信息 */
    memcpy(gossip->ip,n->ip,sizeof(n->ip));
    /* 将被选中节点的端口号记录到 gossip 信息 */
    gossip->port = htons(n->port);
    gossip->cport = htons(n->cport);
    /* 将被选中节点的标识值记录到 gossip 信息 */
    gossip->flags = htons(n->flags);
    gossip->notused1 = 0;
}

下面是 clusterBuildMessageHdr 函数，它主要负责填充消息结构体中的基础信息和当前节点的状态信息。

/* 构建消息的 header */
void clusterBuildMessageHdr(clusterMsg *hdr, int type) {
    int totlen = 0;
    uint64_t offset;
    clusterNode *master;

    /* 如果当前节点是salve，则master为其主节点，如果当前节点是master节点，则master就是当前节点 */
    master = (nodeIsSlave(myself) && myself->slaveof) ?
              myself->slaveof : myself;

    memset(hdr,0,sizeof(*hdr));
    /* 初始化协议版本、标识、及类型， */
    hdr->ver = htons(CLUSTER_PROTO_VER);
    hdr->sig[0] = 'R';
    hdr->sig[1] = 'C';
    hdr->sig[2] = 'm';
    hdr->sig[3] = 'b';
    hdr->type = htons(type);
    /* 消息头设置当前节点id */
    memcpy(hdr->sender,myself->name,CLUSTER_NAMELEN);

    /* 消息头设置当前节点ip */
    memset(hdr->myip,0,NET_IP_STR_LEN);
    if (server.cluster_announce_ip) {
        strncpy(hdr->myip,server.cluster_announce_ip,NET_IP_STR_LEN);
        hdr->myip[NET_IP_STR_LEN-1] = '\0';
    }

    /* 基础端口及集群内节点通信端口 */
    int announced_port = server.cluster_announce_port ?
                         server.cluster_announce_port : server.port;
    int announced_cport = server.cluster_announce_bus_port ?
                          server.cluster_announce_bus_port :
                          (server.port + CLUSTER_PORT_INCR);
    /* 设置当前节点的槽信息 */
    memcpy(hdr->myslots,master->slots,sizeof(hdr->myslots));
    memset(hdr->slaveof,0,CLUSTER_NAMELEN);
    if (myself->slaveof != NULL)
        memcpy(hdr->slaveof,myself->slaveof->name, CLUSTER_NAMELEN);
    hdr->port = htons(announced_port);
    hdr->cport = htons(announced_cport);
    hdr->flags = htons(myself->flags);
    hdr->state = server.cluster->state;

    /* 设置 currentEpoch and configEpochs. */
    hdr->currentEpoch = htonu64(server.cluster->currentEpoch);
    hdr->configEpoch = htonu64(master->configEpoch);

    /* 设置复制偏移量 */
    if (nodeIsSlave(myself))
        offset = replicationGetSlaveOffset();
    else
        offset = server.master_repl_offset;
    hdr->offset = htonu64(offset);

    /* Set the message flags. */
    if (nodeIsMaster(myself) && server.cluster->mf_end)
        hdr->mflags[0] |= CLUSTERMSG_FLAG0_PAUSED;

    /* 计算并设置消息的总长度 */
    if (type == CLUSTERMSG_TYPE_FAIL) {
        totlen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
        totlen += sizeof(clusterMsgDataFail);
    } else if (type == CLUSTERMSG_TYPE_UPDATE) {
        totlen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
        totlen += sizeof(clusterMsgDataUpdate);
    }
    hdr->totlen = htonl(totlen);
}