溫馨提示×
您好，登錄后才能下訂單哦！
點(diǎn)擊 登錄注冊 即表示同意《億速云用戶服務(wù)條款》
您好，登錄后才能下訂單哦！
這篇文章主要講解了“如何理解Redis Cluster Gossip 協(xié)議”，文中的講解內(nèi)容簡單清晰，易于學(xué)習(xí)與理解，下面請大家跟著小編的思路慢慢深入，一起來研究和學(xué)習(xí)“如何理解Redis Cluster Gossip 協(xié)議”吧！
對于數(shù)據(jù)存儲(chǔ)領(lǐng)域，當(dāng)數(shù)據(jù)量或者請求流量大到一定程度后，就必然會(huì)引入分布式。比如 Redis，雖然其單機(jī)性能十分優(yōu)秀，但是因?yàn)橄铝性驎r(shí)，也不得不引入集群。
有集群，集群往往需要維護(hù)一定的元數(shù)據(jù)，比如實(shí)例的ip地址，緩存分片的 slots 信息等，所以需要一套分布式機(jī)制來維護(hù)元數(shù)據(jù)的一致性。這類機(jī)制一般有兩個(gè)模式：分散式和集中式
分散式機(jī)制將元數(shù)據(jù)存儲(chǔ)在部分或者所有節(jié)點(diǎn)上，不同節(jié)點(diǎn)之間進(jìn)行不斷的通信來維護(hù)元數(shù)據(jù)的變更和一致性。Redis Cluster，Consul 等都是該模式。
| 模式 | 優(yōu)點(diǎn) | 缺點(diǎn) |
|---|---|---|
| 集中式 | 數(shù)據(jù)更新及時(shí)，時(shí)效好，元數(shù)據(jù)的更新和讀取，時(shí)效性非常好，一旦元數(shù)據(jù)出現(xiàn)了變更，立即就更新到集中式的外部節(jié)點(diǎn)中，其他節(jié)點(diǎn)讀取的時(shí)候立即就可以感知到; | 較大數(shù)據(jù)更新壓力，更新壓力全部集中在外部節(jié)點(diǎn)，作為單點(diǎn)影響整個(gè)系統(tǒng) |
| 分散式 | 數(shù)據(jù)更新壓力分散，元數(shù)據(jù)的更新比較分散，不是集中某一個(gè)節(jié)點(diǎn)，更新請求比較分散，而且有不同節(jié)點(diǎn)處理，有一定的延時(shí)，降低了并發(fā)壓力 | 數(shù)據(jù)更新延遲，可能導(dǎo)致集群的感知有一定的滯后 |
分散式的元數(shù)據(jù)模式有多種可選的算法進(jìn)行元數(shù)據(jù)的同步，比如說 Paxos、Raft 和 Gossip。Paxos 和 Raft 等都需要全部節(jié)點(diǎn)或者大多數(shù)節(jié)點(diǎn)(超過一半)正常運(yùn)行，整個(gè)集群才能穩(wěn)定運(yùn)行，而 Gossip 則不需要半數(shù)以上的節(jié)點(diǎn)運(yùn)行。
Gossip 協(xié)議，顧名思義，就像流言蜚語一樣，利用一種隨機(jī)、帶有傳染性的方式，將信息傳播到整個(gè)網(wǎng)絡(luò)中，并在一定時(shí)間內(nèi)，使得系統(tǒng)內(nèi)的所有節(jié)點(diǎn)數(shù)據(jù)一致。對你來說，掌握這個(gè)協(xié)議不僅能很好地理解這種最常用的，實(shí)現(xiàn)最終一致性的算法，也能在后續(xù)工作中得心應(yīng)手地實(shí)現(xiàn)數(shù)據(jù)的最終一致性。
Gossip 協(xié)議又稱 epidemic 協(xié)議（epidemic protocol），是基于流行病傳播方式的節(jié)點(diǎn)或者進(jìn)程之間信息交換的協(xié)議，在P2P網(wǎng)絡(luò)和分布式系統(tǒng)中應(yīng)用廣泛，它的方法論也特別簡單：
在一個(gè)處于有界網(wǎng)絡(luò)的集群里，如果每個(gè)節(jié)點(diǎn)都隨機(jī)與其他節(jié)點(diǎn)交換特定信息，經(jīng)過足夠長的時(shí)間后，集群各個(gè)節(jié)點(diǎn)對該份信息的認(rèn)知終將收斂到一致。
這里的“特定信息”一般就是指集群狀態(tài)、各節(jié)點(diǎn)的狀態(tài)以及其他元數(shù)據(jù)等。Gossip協(xié)議是完全符合 BASE 原則，可以用在任何要求最終一致性的領(lǐng)域，比如分布式存儲(chǔ)和注冊中心。另外，它可以很方便地實(shí)現(xiàn)彈性集群，允許節(jié)點(diǎn)隨時(shí)上下線，提供快捷的失敗檢測和動(dòng)態(tài)負(fù)載均衡等。
此外，Gossip 協(xié)議的最大的好處是，即使集群節(jié)點(diǎn)的數(shù)量增加，每個(gè)節(jié)點(diǎn)的負(fù)載也不會(huì)增加很多，幾乎是恒定的。這就允許 Redis Cluster 或者 Consul 集群管理的節(jié)點(diǎn)規(guī)模能橫向擴(kuò)展到數(shù)千個(gè)。
Redis Cluster 是在 3.0 版本引入集群功能。為了讓讓集群中的每個(gè)實(shí)例都知道其他所有實(shí)例的狀態(tài)信息，Redis 集群規(guī)定各個(gè)實(shí)例之間按照 Gossip 協(xié)議來通信傳遞信息。
Redis Cluster 中的每個(gè)節(jié)點(diǎn)都維護(hù)一份自己視角下的當(dāng)前整個(gè)集群的狀態(tài)，主要包括：
也就是說上面的信息，就是集群中Node相互八卦傳播流言蜚語的內(nèi)容主題，而且比較全面，既有自己的更有別人的，這么一來大家都相互傳，最終信息就全面而且一致了。
Redis Cluster 的節(jié)點(diǎn)之間會(huì)相互發(fā)送多種消息，較為重要的如下所示：
Redis 的源碼中 cluster.h 文件定義了全部的消息類型，代碼為 redis 4.0版本。
// 注意，PING 、 PONG 和 MEET 實(shí)際上是同一種消息。
// PONG 是對 PING 的回復(fù)，它的實(shí)際格式也為 PING 消息，
// 而 MEET 則是一種特殊的 PING 消息，用于強(qiáng)制消息的接收者將消息的發(fā)送者添加到集群中（如果節(jié)點(diǎn)尚未在節(jié)點(diǎn)列表中的話）
#define CLUSTERMSG_TYPE_PING 0 /* Ping 消息 */
#define CLUSTERMSG_TYPE_PONG 1 /* Pong 用于回復(fù)Ping */
#define CLUSTERMSG_TYPE_MEET 2 /* Meet 請求將某個(gè)節(jié)點(diǎn)添加到集群中 */
#define CLUSTERMSG_TYPE_FAIL 3 /* Fail 將某個(gè)節(jié)點(diǎn)標(biāo)記為 FAIL */
#define CLUSTERMSG_TYPE_PUBLISH 4 /* 通過發(fā)布與訂閱功能廣播消息 */
#define CLUSTERMSG_TYPE_FAILOVER_AUTH_REQUEST 5 /* 請求進(jìn)行故障轉(zhuǎn)移操作，要求消息的接收者通過投票來支持消息的發(fā)送者 */
#define CLUSTERMSG_TYPE_FAILOVER_AUTH_ACK 6 /* 消息的接收者同意向消息的發(fā)送者投票 */
#define CLUSTERMSG_TYPE_UPDATE 7 /* slots 已經(jīng)發(fā)生變化，消息發(fā)送者要求消息接收者進(jìn)行相應(yīng)的更新 */
#define CLUSTERMSG_TYPE_MFSTART 8 /* 為了進(jìn)行手動(dòng)故障轉(zhuǎn)移，暫停各個(gè)客戶端 */
#define CLUSTERMSG_TYPE_COUNT 9 /* 消息總數(shù) */
通過上述這些消息，集群中的每一個(gè)實(shí)例都能獲得其它所有實(shí)例的狀態(tài)信息。這樣一來，即使有新節(jié)點(diǎn)加入、節(jié)點(diǎn)故障、Slot 變更等事件發(fā)生，實(shí)例間也可以通過 PING、PONG 消息的傳遞，完成集群狀態(tài)在每個(gè)實(shí)例上的同步。下面，我們依次來看看幾種常見的場景。
Redis Cluster 中的節(jié)點(diǎn)都會(huì)定時(shí)地向其他節(jié)點(diǎn)發(fā)送 PING 消息，來交換各個(gè)節(jié)點(diǎn)狀態(tài)信息，檢查各個(gè)節(jié)點(diǎn)狀態(tài)，包括在線狀態(tài)、疑似下線狀態(tài) PFAIL 和已下線狀態(tài) FAIL。
Redis 集群的定時(shí) PING/PONG 的工作原理可以概括成兩點(diǎn)：
下圖顯示了兩個(gè)實(shí)例間進(jìn)行 PING、PONG 消息傳遞的情況，其中實(shí)例一為發(fā)送節(jié)點(diǎn)，實(shí)例二是接收節(jié)點(diǎn)
Redis Cluster 加入新節(jié)點(diǎn)時(shí)，客戶端需要執(zhí)行 CLUSTER MEET 命令，如下圖所示。
節(jié)點(diǎn)一在執(zhí)行 CLUSTER MEET 命令時(shí)會(huì)首先為新節(jié)點(diǎn)創(chuàng)建一個(gè) clusterNode 數(shù)據(jù)，并將其添加到自己維護(hù)的 clusterState 的 nodes 字典中。有關(guān) clusterState 和 clusterNode 關(guān)系，我們在最后一節(jié)會(huì)有詳盡的示意圖和源碼來講解。
然后節(jié)點(diǎn)一會(huì)根據(jù)據(jù) CLUSTER MEET 命令中的 IP 地址和端口號，向新節(jié)點(diǎn)發(fā)送一條 MEET 消息。新節(jié)點(diǎn)接收到節(jié)點(diǎn)一發(fā)送的MEET消息后，新節(jié)點(diǎn)也會(huì)為節(jié)點(diǎn)一創(chuàng)建一個(gè) clusterNode 結(jié)構(gòu)，并將該結(jié)構(gòu)添加到自己維護(hù)的 clusterState 的 nodes 字典中。
接著，新節(jié)點(diǎn)向節(jié)點(diǎn)一返回一條PONG消息。節(jié)點(diǎn)一接收到節(jié)點(diǎn)B返回的PONG消息后，得知新節(jié)點(diǎn)已經(jīng)成功的接收了自己發(fā)送的MEET消息。
最后，節(jié)點(diǎn)一還會(huì)向新節(jié)點(diǎn)發(fā)送一條 PING 消息。新節(jié)點(diǎn)接收到該條 PING 消息后，可以知道節(jié)點(diǎn)A已經(jīng)成功的接收到了自己返回的P ONG消息，從而完成了新節(jié)點(diǎn)接入的握手操作。
MEET 操作成功之后，節(jié)點(diǎn)一會(huì)通過稍早時(shí)講的定時(shí) PING 機(jī)制將新節(jié)點(diǎn)的信息發(fā)送給集群中的其他節(jié)點(diǎn)，讓其他節(jié)點(diǎn)也與新節(jié)點(diǎn)進(jìn)行握手，最終，經(jīng)過一段時(shí)間后，新節(jié)點(diǎn)會(huì)被集群中的所有節(jié)點(diǎn)認(rèn)識(shí)。
Redis Cluster 中的節(jié)點(diǎn)會(huì)定期檢查已經(jīng)發(fā)送 PING 消息的接收方節(jié)點(diǎn)是否在規(guī)定時(shí)間 ( cluster-node-timeout ) 內(nèi)返回了 PONG 消息，如果沒有則會(huì)將其標(biāo)記為疑似下線狀態(tài)，也就是 PFAIL 狀態(tài)，如下圖所示。
然后，節(jié)點(diǎn)一會(huì)通過 PING 消息，將節(jié)點(diǎn)二處于疑似下線狀態(tài)的信息傳遞給其他節(jié)點(diǎn)，例如節(jié)點(diǎn)三。節(jié)點(diǎn)三接收到節(jié)點(diǎn)一的 PING 消息得知節(jié)點(diǎn)二進(jìn)入 PFAIL 狀態(tài)后，會(huì)在自己維護(hù)的 clusterState 的 nodes 字典中找到節(jié)點(diǎn)二所對應(yīng)的 clusterNode 結(jié)構(gòu)，并將主節(jié)點(diǎn)一的下線報(bào)告添加到 clusterNode 結(jié)構(gòu)的 fail_reports 鏈表中。
隨著時(shí)間的推移，如果節(jié)點(diǎn)十 (舉個(gè)例子) 也因?yàn)?PONG 超時(shí)而認(rèn)為節(jié)點(diǎn)二疑似下線了，并且發(fā)現(xiàn)自己維護(hù)的節(jié)點(diǎn)二的 clusterNode 的 fail_reports 中有半數(shù)以上的主節(jié)點(diǎn)數(shù)量的未過時(shí)的將節(jié)點(diǎn)二標(biāo)記為 PFAIL 狀態(tài)報(bào)告日志，那么節(jié)點(diǎn)十將會(huì)把節(jié)點(diǎn)二將被標(biāo)記為已下線 FAIL 狀態(tài)，并且節(jié)點(diǎn)十會(huì)立刻向集群其他節(jié)點(diǎn)廣播主節(jié)點(diǎn)二已經(jīng)下線的 FAIL 消息，所有收到 FAIL 消息的節(jié)點(diǎn)都會(huì)立即將節(jié)點(diǎn)二狀態(tài)標(biāo)記為已下線。如下圖所示。
需要注意的是，報(bào)告疑似下線記錄是由時(shí)效性的，如果超過 cluster-node-timeout *2 的時(shí)間，這個(gè)報(bào)告就會(huì)被忽略掉，讓節(jié)點(diǎn)二又恢復(fù)成正常狀態(tài)。
綜上，我們了解了 Redis Cluster 在定時(shí) PING/PONG、新節(jié)點(diǎn)上線、節(jié)點(diǎn)疑似下線和真正下線等環(huán)節(jié)的原理和操作流程，下面我們來真正看一下 Redis 在這些環(huán)節(jié)的源碼實(shí)現(xiàn)和具體操作。
首先，我們先來講解一下其中涉及的數(shù)據(jù)結(jié)構(gòu)，也就是上文提到的 ClusterNode 等結(jié)構(gòu)。
每個(gè)節(jié)點(diǎn)都會(huì)維護(hù)一個(gè) clusterState 結(jié)構(gòu)，表示當(dāng)前集群的整體狀態(tài)，它的定義如下所示。
typedef struct clusterState {
clusterNode *myself; /* 當(dāng)前節(jié)點(diǎn)的clusterNode信息 */
....
dict *nodes; /* name到clusterNode的字典 */
....
clusterNode *slots[CLUSTER_SLOTS]; /* slot 和節(jié)點(diǎn)的對應(yīng)關(guān)系*/
....
} clusterState;
它有三個(gè)比較關(guān)鍵的字段，具體示意圖如下所示：
clusterNode 結(jié)構(gòu)保存了一個(gè)節(jié)點(diǎn)的當(dāng)前狀態(tài)，比如節(jié)點(diǎn)的創(chuàng)建時(shí)間、節(jié)點(diǎn)的名字、節(jié)點(diǎn) 當(dāng)前的配置紀(jì)元、節(jié)點(diǎn)的IP地址和端口號等等。除此之外，clusterNode結(jié)構(gòu)的 link 屬性是一個(gè)clusterLink結(jié)構(gòu)，該結(jié)構(gòu)保存了連接節(jié)點(diǎn)所需的有關(guān)信息**，比如**套接字描述符，輸入緩沖區(qū)和輸出緩沖區(qū)。clusterNode 還有一個(gè) fail_report 的列表，用來記錄疑似下線報(bào)告。具體定義如下所示。
typedef struct clusterNode {
mstime_t ctime; /* 創(chuàng)建節(jié)點(diǎn)的時(shí)間 */
char name[CLUSTER_NAMELEN]; /* 節(jié)點(diǎn)的名字 */
int flags; /* 節(jié)點(diǎn)標(biāo)識(shí)，標(biāo)記節(jié)點(diǎn)角色或者狀態(tài)，比如主節(jié)點(diǎn)從節(jié)點(diǎn)或者在線和下線 */
uint64_t configEpoch; /* 當(dāng)前節(jié)點(diǎn)已知的集群統(tǒng)一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; /* 當(dāng)前節(jié)點(diǎn)最后一次向該節(jié)點(diǎn)發(fā)送 PING 消息的時(shí)間 */
mstime_t pong_received; /* 當(dāng)前節(jié)點(diǎn)最后一次收到該節(jié)點(diǎn) PONG 消息的時(shí)間 */
mstime_t fail_time; /* FAIL 標(biāo)志位被設(shè)置的時(shí)間 */
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; /* 當(dāng)前節(jié)點(diǎn)的repl便宜 */
char ip[NET_IP_STR_LEN]; /* 節(jié)點(diǎn)的IP 地址 */
int port; /* 端口 */
int cport; /* 通信端口，一般是端口+1000 */
clusterLink *link; /* 和該節(jié)點(diǎn)的 tcp 連接 */
list *fail_reports; /* 下線記錄列表 */
} clusterNode;
clusterNodeFailReport 是記錄節(jié)點(diǎn)下線報(bào)告的結(jié)構(gòu)體， node 是報(bào)告節(jié)點(diǎn)的信息，而 time 則代表著報(bào)告時(shí)間。
typedef struct clusterNodeFailReport {
struct clusterNode *node; /* 報(bào)告當(dāng)前節(jié)點(diǎn)已經(jīng)下線的節(jié)點(diǎn) */
mstime_t time; /* 報(bào)告時(shí)間 */
} clusterNodeFailReport;
了解了 Reids 節(jié)點(diǎn)維護(hù)的數(shù)據(jù)結(jié)構(gòu)體后，我們再來看節(jié)點(diǎn)進(jìn)行通信的消息結(jié)構(gòu)體。通信消息最外側(cè)的結(jié)構(gòu)體為 clusterMsg，它包括了很多消息記錄信息，包括 RCmb 標(biāo)志位，消息總長度，消息協(xié)議版本，消息類型；它還包括了發(fā)送該消息節(jié)點(diǎn)的記錄信息，比如節(jié)點(diǎn)名稱，節(jié)點(diǎn)負(fù)責(zé)的slot信息，節(jié)點(diǎn)ip和端口等；最后它包含了一個(gè) clusterMsgData 來攜帶具體類型的消息。
typedef struct {
char sig[4]; /* 標(biāo)志位，"RCmb" (Redis Cluster message bus). */
uint32_t totlen; /* 消息總長度 */
uint16_t ver; /* 消息協(xié)議版本 */
uint16_t port; /* 端口 */
uint16_t type; /* 消息類型 */
uint16_t count; /* */
uint64_t currentEpoch; /* 表示本節(jié)點(diǎn)當(dāng)前記錄的整個(gè)集群的統(tǒng)一的epoch，用來決策選舉投票等，與configEpoch不同的是：configEpoch表示的是master節(jié)點(diǎn)的唯一標(biāo)志，currentEpoch是集群的唯一標(biāo)志。 */
uint64_t configEpoch; /* 每個(gè)master節(jié)點(diǎn)都有一個(gè)唯一的configEpoch做標(biāo)志，如果和其他master節(jié)點(diǎn)沖突，會(huì)強(qiáng)制自增使本節(jié)點(diǎn)在集群中唯一 */
uint64_t offset; /* 主從復(fù)制偏移相關(guān)信息，主節(jié)點(diǎn)和從節(jié)點(diǎn)含義不同 */
char sender[CLUSTER_NAMELEN]; /* 發(fā)送節(jié)點(diǎn)的名稱 */
unsigned char myslots[CLUSTER_SLOTS/8]; /* 本節(jié)點(diǎn)負(fù)責(zé)的slots信息,16384/8個(gè)char數(shù)組，一共為16384bit */
char slaveof[CLUSTER_NAMELEN]; /* master信息，假如本節(jié)點(diǎn)是slave節(jié)點(diǎn)的話，協(xié)議帶有master信息 */
char myip[NET_IP_STR_LEN]; /* IP */
char notused1[34]; /* 保留字段 */
uint16_t cport; /* 集群的通信端口 */
uint16_t flags; /* 本節(jié)點(diǎn)當(dāng)前的狀態(tài)，比如 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 是一個(gè) union 結(jié)構(gòu)體，它可以為 PING，MEET，PONG 或者 FAIL 等消息體。其中當(dāng)消息為 PING、MEET 和 PONG 類型時(shí)，ping 字段是被賦值的，而是 FAIL 類型時(shí)，fail 字段是被賦值的。
// 注意這是 union 關(guān)鍵字
union clusterMsgData {
/* PING, MEET 或者 PONG 消息時(shí)，ping 字段被賦值 */
struct {
/* Array of N clusterMsgDataGossip structures */
clusterMsgDataGossip gossip[1];
} ping;
/* FAIL 消息時(shí)，fail 被賦值 */
struct {
clusterMsgDataFail about;
} fail;
// .... 省略 publish 和 update 消息的字段
};
clusterMsgDataGossip 是 PING、PONG 和 MEET 消息的結(jié)構(gòu)體，它會(huì)包括發(fā)送消息節(jié)點(diǎn)維護(hù)的其他節(jié)點(diǎn)信息，也就是上文中 clusterState 中 nodes 字段包含的信息，具體代碼如下所示，你也會(huì)發(fā)現(xiàn)二者的字段是類似的。
typedef struct {
/* 節(jié)點(diǎn)的名字，默認(rèn)是隨機(jī)的，MEET消息發(fā)送并得到回復(fù)后，集群會(huì)為該節(jié)點(diǎn)設(shè)置正式的名稱*/
char nodename[CLUSTER_NAMELEN];
uint32_t ping_sent; /* 發(fā)送節(jié)點(diǎn)最后一次給接收節(jié)點(diǎn)發(fā)送 PING 消息的時(shí)間戳，收到對應(yīng) PONG 回復(fù)后會(huì)被賦值為0 */
uint32_t pong_received; /* 發(fā)送節(jié)點(diǎn)最后一次收到接收節(jié)點(diǎn)發(fā)送 PONG 消息的時(shí)間戳 */
char ip[NET_IP_STR_LEN]; /* IP address last time it was seen */
uint16_t port; /* IP*/
uint16_t cport; /* 端口*/
uint16_t flags; /* 標(biāo)識(shí)*/
uint32_t notused1; /* 對齊字符*/
} clusterMsgDataGossip;
typedef struct {
char nodename[CLUSTER_NAMELEN]; /* 下線節(jié)點(diǎn)的名字 */
} clusterMsgDataFail;
看完了節(jié)點(diǎn)維護(hù)的數(shù)據(jù)結(jié)構(gòu)體和發(fā)送的消息結(jié)構(gòu)體后，我們就來看看 Redis 的具體行為源碼了。
Redis 的 clusterCron 函數(shù)會(huì)被定時(shí)調(diào)用，每被執(zhí)行10次，就會(huì)準(zhǔn)備向隨機(jī)的一個(gè)節(jié)點(diǎn)發(fā)送 PING 消息。
它會(huì)先隨機(jī)的選出 5 個(gè)節(jié)點(diǎn)，然后從中選擇最久沒有與之通信的節(jié)點(diǎn)，調(diào)用 clusterSendPing 函數(shù)發(fā)送類型為 CLUSTERMSG_TYPE_PING 的消息
// cluster.c 文件
// clusterCron() 每執(zhí)行 10 次（至少間隔一秒鐘），就向一個(gè)隨機(jī)節(jié)點(diǎn)發(fā)送 gossip 信息
if (!(iteration % 10)) {
int j;
/* 隨機(jī) 5 個(gè)節(jié)點(diǎn)，選出其中一個(gè) */
for (j = 0; j < 5; j++) {
de = dictGetRandomKey(server.cluster->nodes);
clusterNode *this = dictGetVal(de);
/* 不要 PING 連接斷開的節(jié)點(diǎn)，也不要 PING 最近已經(jīng) PING 過的節(jié)點(diǎn) */
if (this->link == NULL || this->ping_sent != 0) continue;
if (this->flags & (CLUSTER_NODE_MYSELF|CLUSTER_NODE_HANDSHAKE))
continue;
/* 對比 pong_received 字段，選出更長時(shí)間未收到其 PONG 消息的節(jié)點(diǎn)(表示好久沒有接受到該節(jié)點(diǎn)的PONG消息了) */
if (min_pong_node == NULL || min_pong > this->pong_received) {
min_pong_node = this;
min_pong = this->pong_received;
}
}
/* 向最久沒有收到 PONG 回復(fù)的節(jié)點(diǎn)發(fā)送 PING 命令 */
if (min_pong_node) {
serverLog(LL_DEBUG,"Pinging node %.40s", min_pong_node->name);
clusterSendPing(min_pong_node->link, CLUSTERMSG_TYPE_PING);
}
}
clusterSendPing 函數(shù)的具體行為我們后續(xù)再了解，因?yàn)樵摵瘮?shù)在其他環(huán)節(jié)也會(huì)經(jīng)常用到
當(dāng)節(jié)點(diǎn)執(zhí)行 CLUSTER MEET 命令后，會(huì)在自身給新節(jié)點(diǎn)維護(hù)一個(gè) clusterNode 結(jié)構(gòu)體，該結(jié)構(gòu)體的 link 也就是TCP連接字段是 null，表示是新節(jié)點(diǎn)尚未建立連接。
clusterCron 函數(shù)中也會(huì)處理這些未建立連接的新節(jié)點(diǎn)，調(diào)用 createClusterLink 創(chuàng)立連接，然后調(diào)用 clusterSendPing 函數(shù)來發(fā)送 MEET 消息
/* cluster.c clusterCron 函數(shù)部分,為未創(chuàng)建連接的節(jié)點(diǎn)創(chuàng)建連接 */
if (node->link == NULL) {
int fd;
mstime_t old_ping_sent;
clusterLink *link;
/* 和該節(jié)點(diǎn)建立連接 */
fd = anetTcpNonBlockBindConnect(server.neterr, node->ip,
node->cport, NET_FIRST_BIND_ADDR);
/* .... fd 為-1時(shí)的異常處理 */
/* 建立 link */
link = createClusterLink(node);
link->fd = fd;
node->link = link;
aeCreateFileEvent(server.el,link->fd,AE_READABLE,
clusterReadHandler,link);
/* 向新連接的節(jié)點(diǎn)發(fā)送 PING 命令，防止節(jié)點(diǎn)被識(shí)進(jìn)入下線 */
/* 如果節(jié)點(diǎn)被標(biāo)記為 MEET ，那么發(fā)送 MEET 命令，否則發(fā)送 PING 命令 */
old_ping_sent = node->ping_sent;
clusterSendPing(link, node->flags & CLUSTER_NODE_MEET ?
CLUSTERMSG_TYPE_MEET : CLUSTERMSG_TYPE_PING);
/* .... */
/* 如果當(dāng)前節(jié)點(diǎn)（發(fā)送者）沒能收到 MEET 信息的回復(fù)，那么它將不再向目標(biāo)節(jié)點(diǎn)發(fā)送命令。*/
/* 如果接收到回復(fù)的話，那么節(jié)點(diǎn)將不再處于 HANDSHAKE 狀態(tài)，并繼續(xù)向目標(biāo)節(jié)點(diǎn)發(fā)送普通 PING 命令*/
node->flags &= ~CLUSTER_NODE_MEET;
}
防止節(jié)點(diǎn)假超時(shí)和標(biāo)記疑似下線標(biāo)記也是在 clusterCron 函數(shù)中，具體如下所示。它會(huì)檢查當(dāng)前所有的 nodes 節(jié)點(diǎn)列表，如果發(fā)現(xiàn)某個(gè)節(jié)點(diǎn)與自己的最后一個(gè) PONG 通信時(shí)間超過了預(yù)定的閾值的一半時(shí)，為了防止節(jié)點(diǎn)是假超時(shí)，會(huì)主動(dòng)釋放掉與之的 link 連接，然后會(huì)主動(dòng)向它發(fā)送一個(gè) PING 消息。
/* cluster.c clusterCron 函數(shù)部分，遍歷節(jié)點(diǎn)來檢查 fail 的節(jié)點(diǎn)*/
while((de = dictNext(di)) != NULL) {
clusterNode *node = dictGetVal(de);
now = mstime(); /* Use an updated time at every iteration. */
mstime_t delay;
/* 如果等到 PONG 到達(dá)的時(shí)間超過了 node timeout 一半的連接 */
/* 因?yàn)楸M管節(jié)點(diǎn)依然正常，但連接可能已經(jīng)出問題了 */
if (node->link && /* is connected */
now - node->link->ctime >
server.cluster_node_timeout && /* 還未重連 */
node->ping_sent && /* 已經(jīng)發(fā)過ping消息 */
node->pong_received < node->ping_sent && /* 還在等待pong消息 */
/* 等待pong消息超過了 timeout/2 */
now - node->ping_sent > server.cluster_node_timeout/2)
{
/* 釋放連接，下次 clusterCron() 會(huì)自動(dòng)重連 */
freeClusterLink(node->link);
}
/* 如果目前沒有在 PING 節(jié)點(diǎn)*/
/* 并且已經(jīng)有 node timeout 一半的時(shí)間沒有從節(jié)點(diǎn)那里收到 PONG 回復(fù) */
/* 那么向節(jié)點(diǎn)發(fā)送一個(gè) PING ，確保節(jié)點(diǎn)的信息不會(huì)太舊，有可能一直沒有隨機(jī)中 */
if (node->link &&
node->ping_sent == 0 &&
(now - node->pong_received) > server.cluster_node_timeout/2)
{
clusterSendPing(node->link, CLUSTERMSG_TYPE_PING);
continue;
}
/* .... 處理failover和標(biāo)記遺失下線 */
}
如果防止節(jié)點(diǎn)假超時(shí)處理后，節(jié)點(diǎn)依舊未收到目標(biāo)節(jié)點(diǎn)的 PONG 消息，并且時(shí)間已經(jīng)超過了 cluster_node_timeout，那么就將該節(jié)點(diǎn)標(biāo)記為疑似下線狀態(tài)。
/* 如果這是一個(gè)主節(jié)點(diǎn)，并且有一個(gè)從服務(wù)器請求進(jìn)行手動(dòng)故障轉(zhuǎn)移,那么向從服務(wù)器發(fā)送 PING*/
if (server.cluster->mf_end &&
nodeIsMaster(myself) &&
server.cluster->mf_slave == node &&
node->link)
{
clusterSendPing(node->link, CLUSTERMSG_TYPE_PING);
continue;
}
/* 后續(xù)代碼只在節(jié)點(diǎn)發(fā)送了 PING 命令的情況下執(zhí)行*/
if (node->ping_sent == 0) continue;
/* 計(jì)算等待 PONG 回復(fù)的時(shí)長 */
delay = now - node->ping_sent;
/* 等待 PONG 回復(fù)的時(shí)長超過了限制值，將目標(biāo)節(jié)點(diǎn)標(biāo)記為 PFAIL （疑似下線)*/
if (delay > server.cluster_node_timeout) {
/* 超時(shí)了，標(biāo)記為疑似下線 */
if (!(node->flags & (REDIS_NODE_PFAIL|REDIS_NODE_FAIL))) {
redisLog(REDIS_DEBUG,"*** NODE %.40s possibly failing",
node->name);
// 打開疑似下線標(biāo)記
node->flags |= REDIS_NODE_PFAIL;
update_state = 1;
}
}
以下是前方多次調(diào)用過的clusterSendPing()方法的源碼，代碼中有詳細(xì)的注釋，大家可以自行閱讀。主要的操作就是將節(jié)點(diǎn)自身維護(hù)的 clusterState 轉(zhuǎn)換為對應(yīng)的消息結(jié)構(gòu)體，。
/* 向指定節(jié)點(diǎn)發(fā)送一條 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 是用于發(fā)送 gossip 信息的計(jì)數(shù)器
// 每次發(fā)送一條信息時(shí)，程序?qū)?nbsp;freshnodes 的值減一
// 當(dāng) freshnodes 的數(shù)值小于等于 0 時(shí)，程序停止發(fā)送 gossip 信息
// freshnodes 的數(shù)量是節(jié)點(diǎn)目前的 nodes 表中的節(jié)點(diǎn)數(shù)量減去 2
// 這里的 2 指兩個(gè)節(jié)點(diǎn)，一個(gè)是 myself 節(jié)點(diǎn)（也即是發(fā)送信息的這個(gè)節(jié)點(diǎn)）
// 另一個(gè)是接受 gossip 信息的節(jié)點(diǎn)
int freshnodes = dictSize(server.cluster->nodes)-2;
/* 計(jì)算要攜帶多少節(jié)點(diǎn)的信息，最少3個(gè)，最多 1/10 集群總節(jié)點(diǎn)數(shù)量*/
wanted = floor(dictSize(server.cluster->nodes)/10);
if (wanted < 3) wanted = 3;
if (wanted > freshnodes) wanted = freshnodes;
/* .... 省略 totlen 的計(jì)算等*/
/* 如果發(fā)送的信息是 PING ，那么更新最后一次發(fā)送 PING 命令的時(shí)間戳 */
if (link->node && type == CLUSTERMSG_TYPE_PING)
link->node->ping_sent = mstime();
/* 將當(dāng)前節(jié)點(diǎn)的信息（比如名字、地址、端口號、負(fù)責(zé)處理的槽）記錄到消息里面 */
clusterBuildMessageHdr(hdr,type);
/* Populate the gossip fields */
int maxiterations = wanted*3;
/* 每個(gè)節(jié)點(diǎn)有 freshnodes 次發(fā)送 gossip 信息的機(jī)會(huì)
每次向目標(biāo)節(jié)點(diǎn)發(fā)送 2 個(gè)被選中節(jié)點(diǎn)的 gossip 信息（gossipcount 計(jì)數(shù)） */
while(freshnodes > 0 && gossipcount < wanted && maxiterations--) {
/* 從 nodes 字典中隨機(jī)選出一個(gè)節(jié)點(diǎn)（被選中節(jié)點(diǎn)） */
dictEntry *de = dictGetRandomKey(server.cluster->nodes);
clusterNode *this = dictGetVal(de);
/* 以下節(jié)點(diǎn)不能作為被選中節(jié)點(diǎn)：
* Myself:節(jié)點(diǎn)本身。
* PFAIL狀態(tài)的節(jié)點(diǎn)
* 處于 HANDSHAKE 狀態(tài)的節(jié)點(diǎn)。
* 帶有 NOADDR 標(biāo)識(shí)的節(jié)點(diǎn)
* 因?yàn)椴惶幚砣魏?nbsp;Slot 而被斷開連接的節(jié)點(diǎn)
*/
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;
}
// 檢查被選中節(jié)點(diǎn)是否已經(jīng)在 hdr->data.ping.gossip 數(shù)組里面
// 如果是的話說明這個(gè)節(jié)點(diǎn)之前已經(jīng)被選中了
// 不要再選中它（否則就會(huì)出現(xiàn)重復(fù)）
if (clusterNodeIsInGossipSection(hdr,gossipcount,this)) continue;
/* 這個(gè)被選中節(jié)點(diǎn)有效，計(jì)數(shù)器減一 */
clusterSetGossipEntry(hdr,gossipcount,this);
freshnodes--;
gossipcount++;
}
/* .... 如果有 PFAIL 節(jié)點(diǎn)，最后添加 */
/* 計(jì)算信息長度 */
totlen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
totlen += (sizeof(clusterMsgDataGossip)*gossipcount);
/* 將被選中節(jié)點(diǎn)的數(shù)量（gossip 信息中包含了多少個(gè)節(jié)點(diǎn)的信息）記錄在 count 屬性里面*/
hdr->count = htons(gossipcount);
/* 將信息的長度記錄到信息里面 */
hdr->totlen = htonl(totlen);
/* 發(fā)送網(wǎng)絡(luò)請求 */
clusterSendMessage(link,buf,totlen);
zfree(buf);
}
void clusterSetGossipEntry(clusterMsg *hdr, int i, clusterNode *n) {
clusterMsgDataGossip *gossip;
/* 指向 gossip 信息結(jié)構(gòu) */
gossip = &(hdr->data.ping.gossip[i]);
/* 將被選中節(jié)點(diǎn)的名字記錄到 gossip 信息 */
memcpy(gossip->nodename,n->name,CLUSTER_NAMELEN);
/* 將被選中節(jié)點(diǎn)的 PING 命令發(fā)送時(shí)間戳記錄到 gossip 信息 */
gossip->ping_sent = htonl(n->ping_sent/1000);
/* 將被選中節(jié)點(diǎn)的 PONG 命令回復(fù)的時(shí)間戳記錄到 gossip 信息 */
gossip->pong_received = htonl(n->pong_received/1000);
/* 將被選中節(jié)點(diǎn)的 IP 記錄到 gossip 信息 */
memcpy(gossip->ip,n->ip,sizeof(n->ip));
/* 將被選中節(jié)點(diǎn)的端口號記錄到 gossip 信息 */
gossip->port = htons(n->port);
gossip->cport = htons(n->cport);
/* 將被選中節(jié)點(diǎn)的標(biāo)識(shí)值記錄到 gossip 信息 */
gossip->flags = htons(n->flags);
gossip->notused1 = 0;
}
下面是 clusterBuildMessageHdr 函數(shù)，它主要負(fù)責(zé)填充消息結(jié)構(gòu)體中的基礎(chǔ)信息和當(dāng)前節(jié)點(diǎn)的狀態(tài)信息。
/* 構(gòu)建消息的 header */
void clusterBuildMessageHdr(clusterMsg *hdr, int type) {
int totlen = 0;
uint64_t offset;
clusterNode *master;
/* 如果當(dāng)前節(jié)點(diǎn)是salve，則master為其主節(jié)點(diǎn)，如果當(dāng)前節(jié)點(diǎn)是master節(jié)點(diǎn)，則master就是當(dāng)前節(jié)點(diǎn) */
master = (nodeIsSlave(myself) && myself->slaveof) ?
myself->slaveof : myself;
memset(hdr,0,sizeof(*hdr));
/* 初始化協(xié)議版本、標(biāo)識(shí)、及類型， */
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);
/* 消息頭設(shè)置當(dāng)前節(jié)點(diǎn)id */
memcpy(hdr->sender,myself->name,CLUSTER_NAMELEN);
/* 消息頭設(shè)置當(dāng)前節(jié)點(diǎn)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';
}
/* 基礎(chǔ)端口及集群內(nèi)節(jié)點(diǎn)通信端口 */
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);
/* 設(shè)置當(dāng)前節(jié)點(diǎn)的槽信息 */
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;
/* 設(shè)置 currentEpoch and configEpochs. */
hdr->currentEpoch = htonu64(server.cluster->currentEpoch);
hdr->configEpoch = htonu64(master->configEpoch);
/* 設(shè)置復(fù)制偏移量 */
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;
/* 計(jì)算并設(shè)置消息的總長度 */
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);
}
感謝各位的閱讀，以上就是“如何理解Redis Cluster Gossip 協(xié)議”的內(nèi)容了，經(jīng)過本文的學(xué)習(xí)后，相信大家對如何理解Redis Cluster Gossip 協(xié)議這一問題有了更深刻的體會(huì)，具體使用情況還需要大家實(shí)踐驗(yàn)證。這里是億速云，小編將為大家推送更多相關(guān)知識(shí)點(diǎn)的文章，歡迎關(guān)注！
免責(zé)聲明：本站發(fā)布的內(nèi)容（圖片、視頻和文字）以原創(chuàng)、轉(zhuǎn)載和分享為主，文章觀點(diǎn)不代表本網(wǎng)站立場，如果涉及侵權(quán)請聯(lián)系站長郵箱：is@yisu.com進(jìn)行舉報(bào)，并提供相關(guān)證據(jù)，一經(jīng)查實(shí)，將立刻刪除涉嫌侵權(quán)內(nèi)容。
億速云公眾號
手機(jī)網(wǎng)站二維碼