二十redis之gossip协议

忘是亡心i 2022-12-10 02:25 228阅读 0赞

二十redis之gossip协议

gossip协议是p2p方式的通信协议。通过节点之间不断交换信息，一段时间后所有节点都会知道整个集群完整的信息。

gossip算法，意思是八卦算法，在办公室中只要一个人八卦一下，在有限的时间内，办公室内的所有人都会知道八卦消息。
算法过程：集群中的一个节点广播自身信息，部分节点收到了信息，这些节点再继续在集群中传播这个节点的信息，一段时间后整个集群中都
有了这个节点的信息。实际上是gossip大部分节点都在一直做这个操作，所以集群在一段时间后信息透明。

通信过程

  1. 每一个节点有两个TCP端口：一个client访问的端口，一个节点间通信端口，通信端口号等于client访问端口加10000
  2. 每个节点在固定周期内通过特定规则选择几个节点发送ping消息。
  3. 接受到ping消息的节点会用Pong消息作为响应。

协议

消息类型分为: ping, pong, meet, fail

gossip协议消息由 消息头+ 消息体组成。

消息头：

  1. typedef struct { 
  2.     char sig[4];        // 消息标识 RCmb
  3.     uint32_t totlen;    // 消息的总长度
  4.     uint16_t ver;       // 协议版本 当前是 1
  5.     uint16_t port;      // 基础端口号 client与server之间通信的端口
  6.     uint16_t type;      // 消息类型
  7.     uint16_t count;     // 如果是ping,pong表示消息体中的节点数
  8.     uint64_t currentEpoch;  //当前发送节点的配置纪元
  9.     uint64_t configEpoch;   // 主节点/从节点的主节点配置纪元
  10.     uint64_t offset;   // 复制偏移量
  11.     char sender[CLUSTER_NAMELEN];// 当前发送节点的nodeId
  12.     unsigned char myslots[CLUSTER_SLOTS/8]; // 当前节点负责的槽信息
  13.     char slaveof[CLUSTER_NAMELEN]; //如果发送节点是从节点，记录对应主节点的nodeId
  14.     char myip[NET_IP_STR_LEN];   // 当前节点的ip
  15.     char notused1[34];  /// 
  16.     uint16_t cport;      //集群节点间通信端口
  17.     uint16_t flags;      // 发送节点标识 区分主从、是否下线
  18.     unsigned char state; // 发送节点所处的结群状态
  19.     unsigned char mflags[3]; // 消息标识
  20.     union clusterMsgData data; // 消息体
  21. } clusterMsg;

消息类型

  1. #define CLUSTERMSG_TYPE_PING 0 /* Ping */
  2. #define CLUSTERMSG_TYPE_PONG 1 /* Pong (reply to Ping) */
  3. #define CLUSTERMSG_TYPE_MEET 2 /* Meet "let's join" message */
  4. #define CLUSTERMSG_TYPE_FAIL 3 /* Mark node xxx as failing */

消息体：

  1. union clusterMsgData { 
  2.     /* PING, MEET and PONG */
  3.     struct { 
  4.         // 数组类型携带多个节点的信息
  5.         clusterMsgDataGossip gossip[1];
  6.     } ping;
  8.     // 失败节点信息
  9.     struct { 
  10.         clusterMsgDataFail about;
  11.     } fail;
  13.     /* PUBLISH */
  14.     struct { 
  15.         clusterMsgDataPublish msg;
  16.     } publish;
  18.     /* UPDATE */
  19.     struct { 
  20.         clusterMsgDataUpdate nodecfg;
  21.     } update;
  22. };

节点Ping，pong消息体结构

  1. typedef struct { 
  2.     char nodename[CLUSTER_NAMELEN]; // nodeId
  3.     uint32_t ping_sent; // 最近一次Ping消息时间
  4.     uint32_t pong_received; //最近一次收到Pong时间
  5.     char ip[NET_IP_STR_LEN]; // node的ip
  6.     uint16_t port;            //node的基础端口
  7.     uint16_t cport;          //node集群间节点通信端口
  8.     uint16_t flags;          //节点标识
  9.     uint32_t notused1;
  10. } clusterMsgDataGossip;

节点fail消息体结构

  1. typedef struct { 
  2.     char nodename[CLUSTER_NAMELEN]; // 失败nodeId
  3. } clusterMsgDataFail;

ping —> pong 节点间通信

ping消息封装了自身和部分其他节点的状态数据

  1. //方法调用链
  3. clusterSendPing() -->  clusterSetGossipEntry()  --> clusterSendMessage()
clusterSendPing
  1. void clusterSendPing(clusterLink *link, int type) { 
  2.     unsigned char *buf;
  3.     // 发送的消息
  4.     clusterMsg *hdr;
  5.     // 消息体中包含的节点数
  6.     int gossipcount = 0;
  7.     // 接受通知的节点数量 
  8.     int wanted; 
  9.     // 消息总长度
  10.     int totlen; 
  12.     int freshnodes = dictSize(server.cluster->nodes)-2;
  14.     // 取集群中的10%的节点
  15.     wanted = floor(dictSize(server.cluster->nodes)/10);
  16.     if (wanted < 3) wanted = 3;
  17.     if (wanted > freshnodes) wanted = freshnodes;
  19.     /* Include all the nodes in PFAIL state, so that failure reports are * faster to propagate to go from PFAIL to FAIL state. */
  20.     int pfail_wanted = server.cluster->stats_pfail_nodes;
  22.     /* Compute the maxium totlen to allocate our buffer. We'll fix the totlen * later according to the number of gossip sections we really were able * to put inside the packet. */
  23.     totlen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
  24.     totlen += (sizeof(clusterMsgDataGossip)*(wanted+pfail_wanted));
  25.     /* Note: clusterBuildMessageHdr() expects the buffer to be always at least * sizeof(clusterMsg) or more. */
  26.     if (totlen < (int)sizeof(clusterMsg)) totlen = sizeof(clusterMsg);
  27.     buf = zcalloc(totlen);
  28.     hdr = (clusterMsg*) buf;
  30.     /* Populate the header. */
  31.     if (link->node && type == CLUSTERMSG_TYPE_PING)
  32.         link->node->ping_sent = mstime();
  33.     // 初始化消息头
  34.     clusterBuildMessageHdr(hdr,type);
  36.     /* Populate the gossip fields */
  37.     int maxiterations = wanted*3;
  38.     while(freshnodes > 0 && gossipcount < wanted && maxiterations--) { 
  39.         // 随机选一个节点
  40.         dictEntry *de = dictGetRandomKey(server.cluster->nodes);
  41.         clusterNode *this = dictGetVal(de);
  43.         /* Don't include this node: the whole packet header is about us * already, so we just gossip about other nodes. */
  44.          // 因为消息头已经包含当前节点信息，所以消息体就不需要了
  45.         if (this == myself) continue;
  47.         /* PFAIL nodes will be added later. */
  48.         //如果节点是失败状态，则进行ping-pong
  49.         if (this->flags & CLUSTER_NODE_PFAIL) continue;
  51.         /* In the gossip section don't include: * 1) Nodes in HANDSHAKE state. * 3) Nodes with the NOADDR flag set. * 4) Disconnected nodes if they don't have configured slots. */
  52.         if (this->flags & (CLUSTER_NODE_HANDSHAKE|CLUSTER_NODE_NOADDR) ||
  53.             (this->link == NULL && this->numslots == 0))
  54.         { 
  55.             freshnodes--; /* Tecnically not correct, but saves CPU. */
  56.             continue;
  57.         }
  59.         /* Do not add a node we already have. */
  60.         if (clusterNodeIsInGossipSection(hdr,gossipcount,this)) continue;
  62.         // 添加节点到消息体中
  63.         clusterSetGossipEntry(hdr,gossipcount,this);
  64.         freshnodes--;
  65.         gossipcount++;
  66.     }
  68.     /* If there are PFAIL nodes, add them at the end. */
  69.     if (pfail_wanted) { 
  70.         dictIterator *di;
  71.         dictEntry *de;
  73.         di = dictGetSafeIterator(server.cluster->nodes);
  74.         while((de = dictNext(di)) != NULL && pfail_wanted > 0) { 
  75.             clusterNode *node = dictGetVal(de);
  76.             if (node->flags & CLUSTER_NODE_HANDSHAKE) continue;
  77.             if (node->flags & CLUSTER_NODE_NOADDR) continue;
  78.             if (!(node->flags & CLUSTER_NODE_PFAIL)) continue;
  79.             clusterSetGossipEntry(hdr,gossipcount,node);
  80.             freshnodes--;
  81.             gossipcount++;
  82.             /* We take the count of the slots we allocated, since the * PFAIL stats may not match perfectly with the current number * of PFAIL nodes. */
  83.             pfail_wanted--;
  84.         }
  85.         dictReleaseIterator(di);
  86.     }
  88.     /* Ready to send... fix the totlen fiend and queue the message in the * output buffer. */
  89.     totlen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
  90.     totlen += (sizeof(clusterMsgDataGossip)*gossipcount);
  91.     hdr->count = htons(gossipcount);
  92.     hdr->totlen = htonl(totlen);
  93.     clusterSendMessage(link,buf,totlen);
  94.     zfree(buf);
  95. }
  98. #define CLUSTER_BROADCAST_ALL 0
  99. #define CLUSTER_BROADCAST_LOCAL_SLAVES 1
  100. void clusterBroadcastPong(int target) { 
  101.     dictIterator *di;
  102.     dictEntry *de;
  104.     di = dictGetSafeIterator(server.cluster->nodes);
  105.     while((de = dictNext(di)) != NULL) { 
  106.         clusterNode *node = dictGetVal(de);
  108.         if (!node->link) continue;
  109.         if (node == myself || nodeInHandshake(node)) continue;
  110.         if (target == CLUSTER_BROADCAST_LOCAL_SLAVES) { 
  111.             int local_slave =
  112.                 nodeIsSlave(node) && node->slaveof &&
  113.                 (node->slaveof == myself || node->slaveof == myself->slaveof);
  114.             if (!local_slave) continue;
  115.         }
  116.         clusterSendPing(node->link,CLUSTERMSG_TYPE_PONG);
  117.     }
  118.     dictReleaseIterator(di);
  119. }
clusterSetGossipEntry
  1. // 构造消息体的单个节点信息
  2. void clusterSetGossipEntry(clusterMsg *hdr, int i, clusterNode *n) { 
  3.     clusterMsgDataGossip *gossip;
  4.     // 节点加入到消息体中的第i个
  5.     gossip = &(hdr->data.ping.gossip[i]);
  6.     memcpy(gossip->nodename,n->name,CLUSTER_NAMELEN);
  7.     gossip->ping_sent = htonl(n->ping_sent/1000);
  8.     gossip->pong_received = htonl(n->pong_received/1000);
  9.     memcpy(gossip->ip,n->ip,sizeof(n->ip));
  10.     gossip->port = htons(n->port);
  11.     gossip->cport = htons(n->cport);
  12.     gossip->flags = htons(n->flags);
  13.     gossip->notused1 = 0;
  14. }

meet —> pong 新节点加入

新加入节点发送meet消息到集群内任一节点(节点B)，通知有新节点加入，节点B加入新节点到自身保存的节点信息，节点B与集群内的节点进行ping-ping通信，
最终集群内的所有节点都保存了新的节点信息。

pong —> other node 广播自身节点信息

pong消息封装了节点自身状态信息。当收到meet,ping消息，作为响应回复给发送方确认通信正常。节点也可以向集群内广播pong消息来通知整个集群对自身状态进行更新。

fial —> other node 广播节点失败

当节点判定集群内另一个节点下线，会向集群内广播一个fail消息，其他节点收到fail消息，会把对应的节点更新为下线状态。

  1. // 方法调用链:
  3. markNodeAsFailingIfNeeded（） ---> clusterSendFail() -->  clusterBuildMessageHdr()  --> clusterBroadcastMessage() --> clusterSendMessage()
  6. markNodeAsFailingIfNeeded -->  clusterNodeFailureReportsCount -->  clusterNodeCleanupFailureReports
markNodeAsFailingIfNeeded
  1. void markNodeAsFailingIfNeeded(clusterNode *node) { 
  2.     // 集群内投票node失败的票数
  3.     int failures;
  4.     // 判定node失败下线需要的票数
  5.     int needed_quorum = (server.cluster->size / 2) + 1;
  6.     // 如果超时时间未到，则不处理
  7.     if (!nodeTimedOut(node)) return; 
  8.     // 如果节点已经判定失败，则不处理
  9.     // #define nodeFailed(n) ((n)->flags & CLUSTER_NODE_FAIL)
  10.     if (nodeFailed(node)) return; 
  11.     // 集群内投票node失败的票数
  12.     failures = clusterNodeFailureReportsCount(node);
  13.      // 如果当前节点是主节点，参与投票
  14.     if (nodeIsMaster(myself)) failures++;
  15.     // 如果票数满足，则接下来发送fail通知集群内其他节点
  16.     if (failures < needed_quorum) return; 
  18.     serverLog(LL_NOTICE,
  19.         "Marking node %.40s as failing (quorum reached).", node->name);
  21.     // 设置节点失败状态
  22.     node->flags &= ~CLUSTER_NODE_PFAIL;
  23.     node->flags |= CLUSTER_NODE_FAIL;
  24.     node->fail_time = mstime();
  26.     // 如果当前节点是主节点，则在集群内广播node失败的消息
  27.     if (nodeIsMaster(myself)) clusterSendFail(node->name);
  28.     clusterDoBeforeSleep(CLUSTER_TODO_UPDATE_STATE|CLUSTER_TODO_SAVE_CONFIG);
  29. }
clusterNodeFailureReportsCount
  1. int clusterNodeFailureReportsCount(clusterNode *node) { 
  2.     // 清理无效的失败票数
  3.     clusterNodeCleanupFailureReports(node);
  4.     // 认为node失败的票数
  5.     return listLength(node->fail_reports);
  6. }
  9. /** ** 当投票失败时间大于maxtime，则认为失败投票无效 **/
  10. void clusterNodeCleanupFailureReports(clusterNode *node) { 
  11.     // 投票认为node失败的节点集合
  12.     list *l = node->fail_reports;
  13.     listNode *ln;
  14.     listIter li;
  15.     clusterNodeFailReport *fr;
  16.     // 最大时间 = 节点超时时间 * 2
  17.     mstime_t maxtime = server.cluster_node_timeout *
  18.                      CLUSTER_FAIL_REPORT_VALIDITY_MULT;
  19.     mstime_t now = mstime();
  21.     listRewind(l,&li);
  22.     while ((ln = listNext(&li)) != NULL) { 
  23.         fr = ln->value;
  24.         // 节点移除，票数减一
  25.         if (now - fr->time > maxtime) listDelNode(l,ln);
  26.     }
  27. }
clusterSendFail
  1. // 节点Nodename失败，通知集群内的其他节点
  2. void clusterSendFail(char *nodename) { 
  3.     unsigned char buf[sizeof(clusterMsg)];
  4.     clusterMsg *hdr = (clusterMsg*) buf;
  5.     // 构造消息头
  6.     clusterBuildMessageHdr(hdr,CLUSTERMSG_TYPE_FAIL);
  7.     //设置消息体fail的节点id
  8.     memcpy(hdr->data.fail.about.nodename,nodename,CLUSTER_NAMELEN);
  9.     // 通知集群内的部分节点
  10.     clusterBroadcastMessage(buf,ntohl(hdr->totlen));
  11. }
clusterBuildMessageHdr
  1. void clusterBuildMessageHdr(clusterMsg *hdr, int type) { 
  2.     int totlen = 0;
  3.     uint64_t offset;
  4.     clusterNode *master;
  6.     // 如果当前节点是salve，则master为其主节点，如果当前节点是master节点，则master为当前
  7.     master = (nodeIsSlave(myself) && myself->slaveof) ?
  8.               myself->slaveof : myself;
  10.     memset(hdr,0,sizeof(*hdr));
  11.     //　初始化协议版本、标识、及类型，
  12.     hdr->ver = htons(CLUSTER_PROTO_VER);
  13.     hdr->sig[0] = 'R';
  14.     hdr->sig[1] = 'C';
  15.     hdr->sig[2] = 'm';
  16.     hdr->sig[3] = 'b';
  17.     hdr->type = htons(type);
  18.     // 消息头设置当前节点id
  19.     memcpy(hdr->sender,myself->name,CLUSTER_NAMELEN);
  21.     // 消息头设置当前节点ip
  22.     memset(hdr->myip,0,NET_IP_STR_LEN);
  23.     if (server.cluster_announce_ip) { 
  24.         strncpy(hdr->myip,server.cluster_announce_ip,NET_IP_STR_LEN);
  25.         hdr->myip[NET_IP_STR_LEN-1] = '\0';
  26.     }
  28.     // 基础端口及集群内节点通信端口
  29.     int announced_port = server.cluster_announce_port ?
  30.                          server.cluster_announce_port : server.port;
  31.     int announced_cport = server.cluster_announce_bus_port ?
  32.                           server.cluster_announce_bus_port :
  33.                           (server.port + CLUSTER_PORT_INCR);
  34.     // 当前节点的槽信息
  35.     memcpy(hdr->myslots,master->slots,sizeof(hdr->myslots));
  36.     memset(hdr->slaveof,0,CLUSTER_NAMELEN);
  37.     if (myself->slaveof != NULL)
  38.         memcpy(hdr->slaveof,myself->slaveof->name, CLUSTER_NAMELEN);
  39.     hdr->port = htons(announced_port);
  40.     hdr->cport = htons(announced_cport);
  41.     hdr->flags = htons(myself->flags);
  42.     hdr->state = server.cluster->state;
  44.     /* Set the currentEpoch and configEpochs. */
  45.     hdr->currentEpoch = htonu64(server.cluster->currentEpoch);
  46.     hdr->configEpoch = htonu64(master->configEpoch);
  48.     // 设置复制偏移量
  49.     if (nodeIsSlave(myself))
  50.         offset = replicationGetSlaveOffset();
  51.     else
  52.         offset = server.master_repl_offset;
  53.     hdr->offset = htonu64(offset);
  55.     /* Set the message flags. */
  56.     if (nodeIsMaster(myself) && server.cluster->mf_end)
  57.         hdr->mflags[0] |= CLUSTERMSG_FLAG0_PAUSED;
  59.     // 计算并设置消息的总长度
  60.     if (type == CLUSTERMSG_TYPE_FAIL) { 
  61.         totlen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
  62.         totlen += sizeof(clusterMsgDataFail);
  63.     } else if (type == CLUSTERMSG_TYPE_UPDATE) { 
  64.         totlen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
  65.         totlen += sizeof(clusterMsgDataUpdate);
  66.     }
  67.     hdr->totlen = htonl(totlen);
  69. }
clusterBroadcastMessage
  1. void clusterBroadcastMessage(void *buf, size_t len) { 
  2.     dictIterator *di;
  3.     dictEntry *de;
  4.     // 集群内节点创建迭代器
  5.     di = dictGetSafeIterator(server.cluster->nodes);
  6.     while((de = dictNext(di)) != NULL) { 
  7.         // 得到一个集群中的一个节点
  8.         clusterNode *node = dictGetVal(de);
  9.         // 是否能发送消息
  10.         if (!node->link) continue;
  11.         if (node->flags & (CLUSTER_NODE_MYSELF|CLUSTER_NODE_HANDSHAKE))
  12.             continue;
  13.         //发送消息
  14.         clusterSendMessage(node->link,buf,len);
  15.     }
  16.     // 释放迭代器
  17.     dictReleaseIterator(di);
  18. }
clusterSendMessage
  1. // 发送消息 link: 接受消息的节点 msg: 消息内容 msglen: 消息长度
  2. void clusterSendMessage(clusterLink *link, unsigned char *msg, size_t msglen) { 
  3.     if (sdslen(link->sndbuf) == 0 && msglen != 0)
  4.         aeCreateFileEvent(server.el,link->fd,AE_WRITABLE,
  5.                     clusterWriteHandler,link);
  7.     link->sndbuf = sdscatlen(link->sndbuf, msg, msglen);
  9.     /* Populate sent messages stats. */
  10.     clusterMsg *hdr = (clusterMsg*) msg;
  11.     uint16_t type = ntohs(hdr->type);
  12.     if (type < CLUSTERMSG_TYPE_COUNT)
  13.         server.cluster->stats_bus_messages_sent[type]++;
  14. }

如何选择部分节点

  1. // 方法调用链
  2. serverCron() --> clusterCron() --> clusterSendPing()

redis保持有一个定时任务，1s运行10次。

  1. 每秒随机5次找出最久没有通信的节点

  2. 最后通信时间大于node_timeout/2

    // 100ms一次,1s运行10次
    run_with_period(100) { 

    1.     if (server.cluster_enabled) clusterCron();
    2. }

    // 每秒随机5次找出最久没有通信的节点
    if (!(iteration % 10)) { 

    1.     int j;
    3.     /* Check a few random nodes and ping the one with the oldest * pong_received time. */
    4.     for (j = 0; j < 5; j++) { 
    5.         de = dictGetRandomKey(server.cluster->nodes);
    6.         clusterNode *this = dictGetVal(de);
    8.         /* Don't ping nodes disconnected or with a ping currently active. */
    9.         if (this->link == NULL || this->ping_sent != 0) continue;
    10.         if (this->flags & (CLUSTER_NODE_MYSELF|CLUSTER_NODE_HANDSHAKE))
    11.             continue;
    12.         if (min_pong_node == NULL || min_pong > this->pong_received) { 
    13.             min_pong_node = this;
    14.             min_pong = this->pong_received;
    15.         }
    16.     }
    17.     if (min_pong_node) { 
    18.         serverLog(LL_DEBUG,"Pinging node %.40s", min_pong_node->name);
    19.         clusterSendPing(min_pong_node->link, CLUSTERMSG_TYPE_PING);
    20.     }
    21. }

    // 最后通信时间大于node_timeout/2
    if (node->link &&

    1.         node->ping_sent == 0 &&
    2.         (now - node->pong_received) > server.cluster_node_timeout/2)
    3.     { 
    4.         clusterSendPing(node->link, CLUSTERMSG_TYPE_PING);
    5.         continue;
    6.     }

每次ping会带最多1/10的节点信息，最少3个节点信息(server.cluster->nodes=6, 三主三从)

  1. //clusterSendPing()方法部分代码段
  3. int freshnodes = dictSize(server.cluster->nodes)-2;
  4. wanted = floor(dictSize(server.cluster->nodes)/10);
  5. if (wanted < 3) wanted = 3;
  6. if (wanted > freshnodes) wanted = freshnodes;
  7. //want数为消息携带的节点信息

为什么要携带1/10的节点信息

How many gossip sections we want to add? 1/10 of the number of nodes
and anyway at least 3. Why 1/10?
If we have N masters, with N/10 entries, and we consider that in
node_timeout we exchange with each other node at least 4 packets
(we ping in the worst case in node_timeout/2 time, and we also
receive two pings from the host), we have a total of 8 packets
in the node_timeout*2 falure reports validity time. So we have
that, for a single PFAIL node, we can expect to receive the following
number of failure reports (in the specified window of time):
PROB * GOSSIP_ENTRIES_PER_PACKET * TOTAL_PACKETS:
PROB = probability of being featured in a single gossip entry,
which is 1 / NUM_OF_NODES.
ENTRIES = 10.
TOTAL_PACKETS = 2 * 4 * NUM_OF_MASTERS.
If we assume we have just masters (so num of nodes and num of masters
is the same), with 1/10 we always get over the majority, and specifically
80% of the number of nodes, to account for many masters failing at the
same time.
Since we have non-voting slaves that lower the probability of an entry
to feature our node, we set the number of entires per packet as
10% of the total nodes we have.

计算我们要附加的gossip节点数,gossip部分的节点数应该是所有节点数的1/10，但是最少应该包含3个节点信息。
之所以在gossip部分需要包含所有节点数的1/10，是为了能够在下线检测时间，也就是2倍的node_timeout时间内，
如果有节点下线的话，能够收到大部分集群节点发来的，关于该节点的下线报告； 1/10这个数是这样来的：
如果共有N个集群节点，在超时时间node_timeout内，当前节点最少会收到其他任一节点发来的4个心跳包：
因节点最长经过node_timeout/2时间，就会其他节点发送一次PING包。节点收到PING包后，会回复PONG包。
因此，在node_timeout时间内，当前节点会收到节点A发来的两个PING包，并且会收到节点A发来的，对于我发过去的PING包的回复包，也就是2个PONG包。
因此，在下线监测时间node_timeout2内，会收到其他任一集群节点发来的8个心跳包。
因此，当前节点总共可以收到8N个心跳包，每个心跳包中，包含下线节点信息的概率是1/10，
因此，收到下线报告的期望值就是8N(1/10)，也就是N*80%，因此，这意味着可以收到大部分节点发来的下线报告。

参考

以上源码基于redis-4.0.6

《redis开发与运维》

https://blog.csdn.net/Jin\_Kwok/article/details/90111631

http://c.biancheng.net/view/375.html

https://www.cnblogs.com/merlindu/p/6417957.html?utm\_source=itdadao&utm\_medium=referral

https://www.jianshu.com/p/652b45591bbf

发表评论

表情：
评论列表 （有 0 条评论，228人围观）

还没有评论，来说两句吧...

相关阅读

    相关 Redis cluster gossip

    目录 一、通信节点选择 1.每0.1秒，如果发现有其他节点连不上，则尝试重连 2.每1秒，从5个随机节点中，选出一个其中最久没有通信的节点，进行ping 3.每0.1秒

    怼烎@怼烎@/ 0 赞/ 67 阅读

    相关 二十redis之gossip协议

    二十redis之gossip协议 gossip协议是p2p方式的通信协议。通过节点之间不断交换信息，一段时间后所有节点都会知道整个集群完整的信息。 > gossip算法，意

    忘是亡心i忘是亡心i/ 0 赞/ 229 阅读

    相关 Gossip协议

    传统的监控，如ceilometer，由于每个节点都会向server报告状态，随着节点数量的增加server的压力随之增大。分布式健康检查可以解决这类性能瓶颈，降节点数量从数百台

    迷南。迷南。/ 0 赞/ 246 阅读

    相关 协议 - Gossip协议

    简单介绍下集群数据同步，集群监控用到的两种常见算法。 > Raft算法 raft 集群中的每个节点都可以根据集群运行的情况在三种状态间切换：follower, cand

    迈不过友情╰迈不过友情╰/ 0 赞/ 421 阅读