redis源碼分析(六)--cluster集羣同步

時間 2019-11-08

原文原文鏈接

Redis集羣消息

做爲支持集羣模式的緩存系統，Redis集羣中的各個節點須要按期地進行通訊，以維持各個節點關於其它節點信息的實時性與一致性。如前一篇文章介紹的，Redis在專用的端口監聽集羣其它節點的鏈接，將集羣內部的的通訊與客戶端的通訊區分開來，任意兩個節點之間創建了兩個tcp鏈接，造成一條全雙工的通道。這篇文章將從集羣消息方面進行介紹，主要介紹消息的格式、種類與不一樣場景下的消息處理。node

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，數據端口與集羣端口tcp

type表明了消息的類型，由宏定義CLUSTERMSG_TYPE_*定義函數

sender是發送消息的節點的nameid，用於查找clusterNode結構ui

myslots是發送節點(發送節點是master)或者發送節點的master節點對應的clustreNode結構中的slots，用於同步各個節點的slots信息this

slaveof，若是發送消息的節點是slave，那麼slaveof存儲它的master的nameidspa

flags表明發送節點的狀態，如該節點是slave仍是master的標誌。code

消息體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命令執行流程以下：

執行meet命令的節點會新建一個clusterNode實例，對應參數指定的ip、port、cport，此時clusterNode實例的flag標記爲CLUSTER_NODE_HANDSHAKE及CLUSTER_NODE_MEET狀態。
執行命令的節點在serverCron中爲該clusterNode新建link，鏈接到指定節點，並向其發送ClUSTERMSG_TYPE_MEET消息，去除clusterNode的CLUSTER_NODE_MEET標記。
接收到ClUSTERMSG_TYPE_MEET消息的節點會爲發送節點創建新的clusterNode實例，此時該clusterNode處於CLUSTER_NODE_HANDSHAKE狀態。
接收方向發送方迴應ClUSTERMSG_TYPE_PONG消息
發送方收到ClUSTERMSG_TYPE_PONG消息，去除對應clusterNode的CLUSTER_NODE_HANDSHAKE標記，並將clusterNode的nameid更新爲pong消息頭的nameid。
接收方在serverCon中主動與發送節點創建鏈接，將link保存到它新創建的clusterNode實例中，併發送ClUSTERMSG_TYPE_PING消息。
發送方收到ClUSTERMSG_TYPE_PING消息，迴應ClUSTERMSG_TYPE_PONG消息
接收方收到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消息的消息體中擴散到其它節點，開啓一個節點離線的判斷流程，具體以下：

nodeA向nodeB發送的ping消息超時未收到響應，將nodeB在nodeA中對應的clusterNode實例的flag標記爲CLUSTER_NODE_PFAIL。
nodeA向其它仍然在線的節點發送ping/pong/meet消息，將nodeB的狀態加入到消息體之中，以一個clusterMsgDataGossip實例表示。
其它節點在收到nodeA的消息後，調用clusterProcessGossipSection處理消息體。
其它節點查找本節點中clusterMsgDataGossip對應的clusterNode實例，並將該結點離線的事件記錄到fail_reports鏈表上。 (這裏假設其它節點認識nodeA，而且nodeA是master，不然CLUSTER_NODE_PFAIL的狀態不會被接受)。
其它節點統計clusterNode實例的fail_reports鏈表上報告該節點離線的master數目，若是數目超過了全部master節點的一半，將其標記爲CLUSTER_NODE_FAIL狀態。此外，若是當前作出如此判斷的節點是master節點，那麼向其它節點發送CLUSTERMSG_TYPE_FAIL消息，消息體中帶有離線節點的nameid。
其它節點收到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除了報告離線節點的做用外，還有如下做用：

若是接收消息的節點查找不到clusterMsgDataGossip實例中的nameid，那麼開啓一段發現新節點的流程。
若是接收消息的節點查找到clusterMsgDataGossip對應的clusterNode處於離線狀態，可是clusterMsgDataGossip中狀態倒是在線，而且兩個結構中記錄的ip、port、cport不一樣，那麼更新本地記錄的ip、port、cport，以便嘗試從新鏈接。

5. 更新slots配置

每個集羣消息都在消息頭攜帶了發送節點的信息，其中slots是發送節點或者其master節點負責的slot的bit掩碼，接收方收到收到消息後將查找本地對應發送節點的clusterNode結構，對比兩份slots掩碼是否有所不一樣。消息頭與clusterNode中都有一個參數configEpoch記錄信息的更新時間，值越大表示配置越新，兩個節點中slots不相同，以新的配置爲準。根據狀況作以下處理：

發送方是主節點，而且消息頭中的slots與本地對應clusterNode中的slots不相同，調用函數clusterUpdateSlotsConfigWith更新clusterState與clusterNode中的slots。
若是消息頭中的slots聲明的slot在接收節點中發現由其它節點clusterNode_x負責，而且clusterNode_x中記錄的configEpoch值更大，那麼向發送方迴應CLUSTERMSG_TYPE_UPDATE消息，通知它更新它的slots配置。
接收到CLUSTERMSG_TYPE_UPDATE消息的節點調用clusterUpdateSlotsConfigWith函數更新它的slots配置。

6. 替換主節點

在serverCron中按期執行函數clusterHandleSlaveFailover，檢查是否須要以本節點替換它的master，當集羣中的master離線，或者客戶端執行命令主動替換master節點，集羣開啓替換流程：

執行替換的節點廣播發送CLUSTERMSG_TYPE_FAILOVER_AUTH_REQUEST類型的消息，請求其它master節點回應它的替換請求。
接收到CLUSTERMSG_TYPE_FAILOVER_AUTH_REQUEST消息的節點調用函數clusterSendFailoverAuthIfNeeded判斷它是否迴應請求。僅有效的master節點，而且判斷髮送方符合條件纔回應CLUSTERMSG_TYPE_FAILOVER_AUTH_ACK消息。
收到CLUSTERMSG_TYPE_FAILOVER_AUTH_ACK消息的節點增長它的受權記錄，在clusterHandleSlaveFailover函數中判斷受權數是否大於全部有效master節點數的一半，是則調用函數clusterFailoverReplaceYourMaster執行替換，不然繼續等待。

一個master節點可能有多個slave，每個slave將根據它數據的完整性決定它發送CLUSTERMSG_TYPE_FAILOVER_AUTH_REQUEST請求前須要等待的時間，數據完整的slave更早發送請求，其它master節點回應了某個slave的請求這個事件會被記錄，它將再也不迴應針對同一個master的替換請求。