Redis源码剖析和注释（二十五）--- Redis Cluster 的通信流程深入剖析（载入配置文件、节点握手、分配槽）

Redis Cluster 通信流程深入剖析

1. Redis Cluster 介绍和搭建

Redis Cluster

2. Redis Cluster 和 Redis Sentinel

Redis 2.8

Redis Cluster(哨兵)

Redis Sentinel(集群)

Redis 3.0

Redis Cluster

Redis Sentinel

Redis

Redis

Redis

Redis Sentinel

Redis Sentinel

Redis Cluster

Redis

Cluster

Redis Sentinel

3. 搭建 Redis Cluster的通信流程深入剖析

Redis Cluster

3.1 准备节点

6

cluster-enabled yes

Redis

sudo redis-server conf/redis-6379.conf

main()

loadServerConfig(configfile,options)

loadServerConfigFromString()

cluster-

initServer()

serverCron()

100ms

clusterCron()

clusterInit()

server.cluster

clusterState

clusterInit()

verifyClusterConfigWithData()

aeSetBeforeSleepProc()

beforeSleep()

clusterBeforeSleep()

aeMain()

beforeSleep()

clusterState

typedef struct clusterState {
clusterNode *myself;  /* This node */
// 当前纪元
uint64_t currentEpoch;
// 集群的状态
int state;            /* CLUSTER_OK, CLUSTER_FAIL, ... */
// 集群中至少负责一个槽的主节点个数
int size;             /* Num of master nodes with at least one slot */
// 保存集群节点的字典，键是节点名字，值是clusterNode结构的指针
dict *nodes;          /* Hash table of name -> clusterNode structures */
// 防止重复添加节点的黑名单
dict *nodes_black_list; /* Nodes we don't re-add for a few seconds. */
// 导入槽数据到目标节点，该数组记录这些节点
clusterNode *migrating_slots_to[CLUSTER_SLOTS];
// 导出槽数据到目标节点，该数组记录这些节点
clusterNode *importing_slots_from[CLUSTER_SLOTS];
// 槽和负责槽节点的映射
clusterNode *slots[CLUSTER_SLOTS];
// 槽映射到键的有序集合
zskiplist *slots_to_keys;
/* The following fields are used to take the slave state on elections. */
// 之前或下一次选举的时间
mstime_t failover_auth_time; /* Time of previous or next election. */
// 节点获得支持的票数
int failover_auth_count;    /* Number of votes received so far. */
// 如果为真，表示本节点已经向其他节点发送了投票请求
int failover_auth_sent;     /* True if we already asked for votes. */
// 该从节点在当前请求中的排名
int failover_auth_rank;     /* This slave rank for current auth request. */
// 当前选举的纪元
uint64_t failover_auth_epoch; /* Epoch of the current election. */
// 从节点不能执行故障转移的原因
int cant_failover_reason;
/* Manual failover state in common. */
// 如果为0，表示没有正在进行手动的故障转移。否则表示手动故障转移的时间限制
mstime_t mf_end;
/* Manual failover state of master. */
// 执行手动孤战转移的从节点
clusterNode *mf_slave;      /* Slave performing the manual failover. */
/* Manual failover state of slave. */
// 从节点记录手动故障转移时的主节点偏移量
long long mf_master_offset;
// 非零值表示手动故障转移能开始
int mf_can_start;
/* The followign fields are used by masters to take state on elections. */
// 集群最近一次投票的纪元
uint64_t lastVoteEpoch;     /* Epoch of the last vote granted. */
// 调用clusterBeforeSleep()所做的一些事
int todo_before_sleep; /* Things to do in clusterBeforeSleep(). */
// 发送的字节数
long long stats_bus_messages_sent;  /* Num of msg sent via cluster bus. */
// 通过Cluster接收到的消息数量
long long stats_bus_messages_received; /* Num of msg rcvd via cluster bus.*/
} clusterState;

myself = server.cluster->myself = createClusterNode(NULL,CLUSTER_NODE_MYSELF|CLUSTER_NODE_MASTER);

myself

cluster.h

server.cluster->myself

createClusterNode()

typedef struct clusterNode {
// 节点创建的时间
mstime_t ctime; /* Node object creation time. */
// 名字
char name[CLUSTER_NAMELEN]; /* Node name, hex string, sha1-size */
// 标识
int flags;      /* CLUSTER_NODE_... */
uint64_t configEpoch; /* Last configEpoch observed for this node */
// 节点的槽位图
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 */
// 指向主节点，即使是从节点也可以为NULL
struct clusterNode *slaveof;
// 最近一次发送PING的时间
mstime_t ping_sent;      /* Unix time we sent latest ping */
// 接收到PONG的时间
mstime_t pong_received;  /* Unix time we received the pong */
// 被设置为FAIL的下线时间
mstime_t fail_time;      /* Unix time when FAIL flag was set */
// 最近一次为从节点投票的时间
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;      /* Last known repl offset for this node. */
// ip地址
char ip[NET_IP_STR_LEN];  /* Latest known IP address of this node */
// 节点端口号
int port;                   /* Latest known port of this node */
// 与该节点关联的连接对象
clusterLink *link;          /* TCP/IP link with this node */
// 保存下线报告的链表
list *fail_reports;         /* List of nodes signaling this as failing */
} clusterNode;

link

clusterLink

typedef struct clusterLink {
// 连接创建的时间
mstime_t ctime;             /* Link creation time */
// TCP连接的文件描述符
int fd;                     /* TCP socket file descriptor */
// 输出（发送）缓冲区
sds sndbuf;                 /* Packet send buffer */
// 输入（接收）缓冲区
sds rcvbuf;                 /* Packet reception buffer */
// 关联该连接的节点
struct clusterNode *node;   /* Node related to this link if any, or NULL */
} clusterLink;

MEET

clusterLink

fd

node

clusterLink

link

3.2 节点握手

Gossip

myself

MEET

MEET

PONG

myself

myself

PONG

PING

1s

PING

5

myself

3.2.1

myself

节点发送 MEET 消息

cluster meet <ip> <port>

cluster meet

clusterCommand()

cluster

meet

// CLUSTER MEET <ip> <port>命令
// 与给定地址的节点建立连接
if (!strcasecmp(c->argv[1]->ptr,"meet") && c->argc == 4) {
long long port;
// 获取端口
if (getLongLongFromObject(c->argv[3], &port) != C_OK) {
addReplyErrorFormat(c,"Invalid TCP port specified: %s",
(char*)c->argv[3]->ptr);
return;
}
// 如果没有正在进行握手，那么根据执行的地址开始进行握手操作
if (clusterStartHandshake(c->argv[2]->ptr,port) == 0 &&
errno == EINVAL)
{
addReplyErrorFormat(c,"Invalid node address specified: %s:%s",
(char*)c->argv[2]->ptr, (char*)c->argv[3]->ptr);
// 连接成功回复ok
} else {
addReply(c,shared.ok);
}
}

cluster meet <ip> <port>

clusterStartHandshake()

int clusterStartHandshake(char *ip, int port) {
clusterNode *n;
char norm_ip[NET_IP_STR_LEN];
struct sockaddr_storage sa;

// 检查地址是否非法
if (inet_pton(AF_INET,ip,
&(((struct sockaddr_in *)&sa)->sin_addr)))
{
sa.ss_family = AF_INET;
} else if (inet_pton(AF_INET6,ip,
&(((struct sockaddr_in6 *)&sa)->sin6_addr)))
{
sa.ss_family = AF_INET6;
} else {
errno = EINVAL;
return 0;
}
// 检查端口号是否合法
if (port <= 0 || port > (65535-CLUSTER_PORT_INCR)) {
errno = EINVAL;
return 0;
}
// 设置 norm_ip 作为节点地址的标准字符串表示形式
memset(norm_ip,0,NET_IP_STR_LEN);
if (sa.ss_family == AF_INET)
inet_ntop(AF_INET,
(void*)&(((struct sockaddr_in *)&sa)->sin_addr),
norm_ip,NET_IP_STR_LEN);
else
inet_ntop(AF_INET6,
(void*)&(((struct sockaddr_in6 *)&sa)->sin6_addr),
norm_ip,NET_IP_STR_LEN);
// 判断当前地址是否处于握手状态，如果是，则设置errno并返回，该函数被用来避免重复和相同地址的节点进行握手
if (clusterHandshakeInProgress(norm_ip,port)) {
errno = EAGAIN;
return 0;
}

// 为node设置一个随机的地址，当握手完成时会为其设置真正的名字
// 创建一个随机名字的节点
n = createClusterNode(NULL,CLUSTER_NODE_HANDSHAKE|CLUSTER_NODE_MEET);
// 设置地址
memcpy(n->ip,norm_ip,sizeof(n->ip));
n->port = port;
// 添加到集群中
clusterAddNode(n);
return 1;
}

errno

clusterHandshakeInProgress()

n = createClusterNode(NULL,CLUSTER_NODE_HANDSHAKE|CLUSTER_NODE_MEET);

clusterAddNode()

server.cluster->nodes

clusterNode()

myself

meet

meet

clusterCron()

MEET

// 获取握手状态超时的时间，最低为1s
// 如果一个处于握手状态的节点如果没有在该超时时限内变成一个普通的节点，那么该节点从节点字典中被删除
handshake_timeout = server.cluster_node_timeout;
if (handshake_timeout < 1000) handshake_timeout = 1000;

// 检查是否当前集群中有断开连接的节点和重新建立连接的节点
di = dictGetSafeIterator(server.cluster->nodes);
// 遍历所有集群中的节点，如果有未建立连接的节点，那么发送PING或PONG消息，建立连接
while((de = dictNext(di)) != NULL) {
clusterNode *node = dictGetVal(de);
// 跳过myself节点和处于NOADDR状态的节点
if (node->flags & (CLUSTER_NODE_MYSELF|CLUSTER_NODE_NOADDR)) continue;

// 如果仍然node节点处于握手状态，但是从建立连接开始到现在已经超时
if (nodeInHandshake(node) && now - node->ctime > handshake_timeout) {
// 从集群中删除该节点，遍历下一个节点
clusterDelNode(node);
continue;
}
// 如果节点的连接对象为空
if (node->link == NULL) {
int fd;
mstime_t old_ping_sent;
clusterLink *link;
// myself节点连接这个node节点
fd = anetTcpNonBlockBindConnect(server.neterr, node->ip,
node->port+CLUSTER_PORT_INCR, NET_FIRST_BIND_ADDR);
// 连接出错，跳过该节点
if (fd == -1) {
// 如果ping_sent为0，察觉故障无法执行，因此要设置发送PING的时间，当建立连接后会真正的的发送PING命令
if (node->ping_sent == 0) node->ping_sent = mstime();
serverLog(LL_DEBUG, "Unable to connect to "
"Cluster Node [%s]:%d -> %s", node->ip,
node->port+CLUSTER_PORT_INCR,
server.neterr);
continue;
}
// 为node节点创建一个连接对象
link = createClusterLink(node);
// 设置连接对象的属性
link->fd = fd;
// 为node设置连接对象
node->link = link;
// 监听该连接的可读事件，设置可读时间的读处理函数
aeCreateFileEvent(server.el,link->fd,AE_READABLE,clusterReadHandler,link);
// 备份旧的发送PING的时间
old_ping_sent = node->ping_sent;
// 如果node节点指定了MEET标识，那么发送MEET命令，否则发送PING命令
clusterSendPing(link, node->flags & CLUSTER_NODE_MEET ?
CLUSTERMSG_TYPE_MEET : CLUSTERMSG_TYPE_PING);
// 如果不是第一次发送PING命令，要将发送PING的时间还原，等待被clusterSendPing()更新
if (old_ping_sent) {
node->ping_sent = old_ping_sent;
}
// 发送MEET消息后，清除MEET标识
// 如果没有接收到PONG回复，那么不会在向该节点发送消息
// 如果接收到了PONG回复，取消MEET/HANDSHAKE状态，发送一个正常的PING消息。
node->flags &= ~CLUSTER_NODE_MEET;
serverLog(LL_DEBUG,"Connecting with Node %.40s at %s:%d",
node->name, node->ip, node->port+CLUSTER_PORT_INCR);
}
}
dictReleaseIterator(di);

clusterNode()

myself

myself

myself

TCP

fd

clusterLink

myself

fd

clusterReadHandler()

MEET

PING

myself

clusterSendPing()

MEET

MEET

PING

PING

CLUSTER_NODE_MEET

CLUSTER_NODE_HANDSHAKE

MEET

3.2.2 目标节点处理 MEET 消息回复 PONG 消息

myself

MEET

clusterReadHandler()

PONG

MEET

PONG

myself

clusterInit

fd

server.cfd

fd

clusterAcceptHandler()

myself

MEET

clusterAcceptHandler()

accept()

myself

clusterReadHandler()

myself

myself

MEET

clusterReadHandler()

MEET

clusterReadHandler()

read()

void clusterReadHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
char buf[sizeof(clusterMsg)];
ssize_t nread;
clusterMsg *hdr;
clusterLink *link = (clusterLink*) privdata;
unsigned int readlen, rcvbuflen;
UNUSED(el);
UNUSED(mask);

// 循环从fd读取数据
while(1) { /* Read as long as there is data to read. */
// 获取连接对象的接收缓冲区的长度，表示一次最多能多大的数据量
rcvbuflen = sdslen(link->rcvbuf);
// 如果接收缓冲区的长度小于八字节，就无法读入消息的总长
if (rcvbuflen < 8) {
readlen = 8 - rcvbuflen;
// 能够读入完整数据信息
} else {
hdr = (clusterMsg*) link->rcvbuf;
// 如果是8个字节
if (rcvbuflen == 8) {
// 如果前四个字节不是"RCmb"签名，释放连接
if (memcmp(hdr->sig,"RCmb",4) != 0 ||
ntohl(hdr->totlen) < CLUSTERMSG_MIN_LEN)
{
serverLog(LL_WARNING,
"Bad message length or signature received "
"from Cluster bus.");
handleLinkIOError(link);
return;
}
}
// 记录已经读入的内容长度
readlen = ntohl(hdr->totlen) - rcvbuflen;
if (readlen > sizeof(buf)) readlen = sizeof(buf);
}
// 从fd中读数据
nread = read(fd,buf,readlen);
// 没有数据可读
if (nread == -1 && errno == EAGAIN) return; /* No more data ready. */
// 读错误，释放连接
if (nread <= 0) {
serverLog(LL_DEBUG,"I/O error reading from node link: %s",
(nread == 0) ? "connection closed" : strerror(errno));
handleLinkIOError(link);
return;
} else {
// 将读到的数据追加到连接对象的接收缓冲区中
link->rcvbuf = sdscatlen(link->rcvbuf,buf,nread);
hdr = (clusterMsg*) link->rcvbuf;
rcvbuflen += nread;
}

// 检查接收的数据是否完整
if (rcvbuflen >= 8 && rcvbuflen == ntohl(hdr->totlen)) {
// 如果读到的数据有效，处理读到接收缓冲区的数据
if (clusterProcessPacket(link)) {
// 处理成功，则设置新的空的接收缓冲区
sdsfree(link->rcvbuf);
link->rcvbuf = sdsempty();
} else {
return; /* Link no longer valid. */
}
}
}
}

clusterLink

link->rcvbuf

link->sndbuf

fd

link

clusterProcessPacket()

MEET

MEET

// 从集群中查找sender节点
sender = clusterLookupNode(hdr->sender);

// 初始处理PING和MEET请求，用PONG作为回复
if (type == CLUSTERMSG_TYPE_PING || type == CLUSTERMSG_TYPE_MEET) {
serverLog(LL_DEBUG,"Ping packet received: %p", (void*)link->node);

// 我们使用传入的MEET消息来设置当前myself节点的地址，因为只有其他集群中的节点在握手的时会发送MEET消息，当有节点加入集群时，或者如果我们改变地址，这些节点将使用我们公开的地址来连接我们，所以在集群中，通过套接字来获取地址是一个简单的方法去发现或更新我们自己的地址，而不是在配置中的硬设置
// 但是，如果我们根本没有地址，即使使用正常的PING数据包，我们也会更新该地址。 如果是错误的，那么会被MEET修改
// 如果是MEET消息
// 或者是其他消息但是当前集群节点的IP为空
if (type == CLUSTERMSG_TYPE_MEET || myself->ip[0] == '\0') {
char ip[NET_IP_STR_LEN];
// 可以根据fd来获取ip，并设置myself节点的IP
if (anetSockName(link->fd,ip,sizeof(ip),NULL) != -1 &&
strcmp(ip,myself->ip))
{
memcpy(myself->ip,ip,NET_IP_STR_LEN);
serverLog(LL_WARNING,"IP address for this node updated to %s",
myself->ip);
clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG);
}
}

// 如果当前sender节点是一个新的节点，并且消息是MEET消息类型，那么将这个节点添加到集群中
// 当前该节点的flags、slaveof等等都没有设置，当从其他节点接收到PONG时可以从中获取到信息
if (!sender && type == CLUSTERMSG_TYPE_MEET) {
clusterNode *node;
// 创建一个处于握手状态的节点
node = createClusterNode(NULL,CLUSTER_NODE_HANDSHAKE);
// 设置ip和port
nodeIp2String(node->ip,link);
node->port = ntohs(hdr->port);
// 添加到集群中
clusterAddNode(node);
clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG);
}

// 如果是从一个未知的节点发送过来MEET包，处理流言信息
if (!sender && type == CLUSTERMSG_TYPE_MEET)
// 处理流言中的 PING or PONG 数据包
clusterProcessGossipSection(hdr,link);

/* Anyway reply with a PONG */
// 回复一个PONG消息
clusterSendPing(link,CLUSTERMSG_TYPE_PONG);
}

MEET

clusterLookupNode()

MEET

hdr->sender

myself

myself

sender

NULL

if

MEET

if

MEET

sender

myself

myself

clusterSendPing()

PONG

myself

3.2.3

myself

节点处理 PONG 消息回复 PING 消息

myself

MEET

fd

MEET

PONG

myself

clusterReadHandler()

PONG

myself

PONG

clusterReadHandler()

myself

MEET

read()

link->rcvbuf

标题3.2.2

clusterProcessPacket()

PONG

myself

myself

if (type == CLUSTERMSG_TYPE_PING || type == CLUSTERMSG_TYPE_PONG ||
type == CLUSTERMSG_TYPE_MEET)
{
serverLog(LL_DEBUG,"%s packet received: %p",
type == CLUSTERMSG_TYPE_PING ? "ping" : "pong",
(void*)link->node);
// 如果关联该连接的节点存在
if (link->node) {
// 如果关联该连接的节点处于握手状态
if (nodeInHandshake(link->node)) {
// sender节点存在，用该新的连接地址更新sender节点的地址
if (sender) {
serverLog(LL_VERBOSE,
"Handshake: we already know node %.40s, "
"updating the address if needed.", sender->name);
if (nodeUpdateAddressIfNeeded(sender,link,ntohs(hdr->port)))
{
clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG|
CLUSTER_TODO_UPDATE_STATE);
}
// 释放关联该连接的节点
clusterDelNode(link->node);
return 0;
}
// 将关联该连接的节点的名字用sender的名字替代
clusterRenameNode(link->node, hdr->sender);
serverLog(LL_DEBUG,"Handshake with node %.40s completed.",
link->node->name);
// 取消握手状态，设置节点的角色
link->node->flags &= ~CLUSTER_NODE_HANDSHAKE;
link->node->flags |= flags&(CLUSTER_NODE_MASTER|CLUSTER_NODE_SLAVE);
clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG);
// 如果sender的地址和关联该连接的节点的地址不相同
} else if (memcmp(link->node->name,hdr->sender,
CLUSTER_NAMELEN) != 0)
{
serverLog(LL_DEBUG,"PONG contains mismatching sender ID. About node %.40s added %d ms ago, having flags %d",
link->node->name,
(int)(mstime()-(link->node->ctime)),
link->node->flags);
// 设置NOADDR标识，情况关联连接节点的地址
link->node->flags |= CLUSTER_NODE_NOADDR;
link->node->ip[0] = '\0';
link->node->port = 0;
// 释放连接对象
freeClusterLink(link);
clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG);
return 0;
}
}

// 关联该连接的节点存在，且消息类型为PONG
if (link->node && type == CLUSTERMSG_TYPE_PONG) {
// 更新接收到PONG的时间
link->node->pong_received = mstime();
// 清零最近一次发送PING的时间戳
link->node->ping_sent = 0;

// 接收到PONG回复，可以删除PFAIL（疑似下线）标识
// FAIL标识能否删除，需要clearNodeFailureIfNeeded()来决定
// 如果关联该连接的节点疑似下线
if (nodeTimedOut(link->node)) {
// 取消PFAIL标识
link->node->flags &= ~CLUSTER_NODE_PFAIL;
clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG|
CLUSTER_TODO_UPDATE_STATE);
// 如果关联该连接的节点已经被判断为下线
} else if (nodeFailed(link->node)) {
// 如果一个节点被标识为FAIL，需要检查是否取消该节点的FAIL标识，因为该节点在一定时间内重新上线了
clearNodeFailureIfNeeded(link->node);
}
}
}

MEET

myself

myself

MEET

myself

if (nodeInHandshake(link->node))

myself

MEET

CLUSTER_NODE_MEET

CLUSTER_NODE_HANDSHAKE

if (sender)

PONG

sender

myself

myself

sender

clusterRenameNode()

myself

CLUSTER_NODE_HANDSHAKE

if (link->node && type == CLUSTERMSG_TYPE_PONG)

PONG

PING

PONG

myself

PING

clusterCron()

while((de = dictNext(di)) != NULL) {
if (node->flags &
(CLUSTER_NODE_MYSELF|CLUSTER_NODE_NOADDR|CLUSTER_NODE_HANDSHAKE))
continue;

if (node->link && node->ping_sent == 0 &&
(now - node->pong_received) > server.cluster_node_timeout/2)
{
// 给node节点发送一个PING消息
clusterSendPing(node->link, CLUSTERMSG_TYPE_PING);
continue;
}
}

myself

myself

myself

PONG

node->ping_sent

0

PING

PING

PING

1s

PING

3.2.4 Gossip协议

Redis Cluster

CLUSTER MEET

CLUSTER MEET

Gossip

Gossip

Redis

PING

PONG

Redis Cluster

PING

PONG

MEET

UPDATE

PUBLISH

FAIL

clusterMsg

union clusterMsgData

PING

PONG

MEET

clusterMsgDataGossip

typedef struct {
// 节点名字
char nodename[CLUSTER_NAMELEN];
// 最近一次发送PING的时间戳
uint32_t ping_sent;
// 最近一次接收PONG的时间戳
uint32_t pong_received;
// 节点的IP地址
char ip[NET_IP_STR_LEN];  /* IP address last time it was seen */
// 节点的端口号
uint16_t port;              /* port last time it was seen */
// 节点的标识
uint16_t flags;             /* node->flags copy */
// 未使用
uint16_t notused1;          /* Some room for future improvements. */
uint32_t notused2;
} clusterMsgDataGossip;

clusterSendPing()

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 的值是除了当前myself节点和发送消息的两个节点之外，集群中的所有节点
// freshnodes 表示的意思是gossip协议中可以包含的有关节点信息的最大个数
int freshnodes = dictSize(server.cluster->nodes)-2;
// wanted 的值是集群节点的十分之一向下取整，并且最小等于3
// wanted 表示的意思是gossip中要包含的其他节点信息个数
wanted = floor(dictSize(server.cluster->nodes)/10);
if (wanted < 3) wanted = 3;
// 因此 wanted 最多等于 freshnodes。
if (wanted > freshnodes) wanted = freshnodes;

// 计算分配消息的最大空间
totlen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
totlen += (sizeof(clusterMsgDataGossip)*wanted);
// 消息的总长最少为一个消息结构的大小
if (totlen < (int)sizeof(clusterMsg)) totlen = sizeof(clusterMsg);
// 分配空间
buf = zcalloc(totlen);
hdr = (clusterMsg*) buf;

// 设置发送PING命令的时间
if (link->node && type == CLUSTERMSG_TYPE_PING)
link->node->ping_sent = mstime();
// 构建消息的头部
clusterBuildMessageHdr(hdr,type);

int maxiterations = wanted*3;
// 构建消息内容
while(freshnodes > 0 && gossipcount < wanted && maxiterations--) {
// 随机选择一个集群节点
dictEntry *de = dictGetRandomKey(server.cluster->nodes);
clusterNode *this = dictGetVal(de);
clusterMsgDataGossip *gossip;
int j;

// 1. 跳过当前节点，不选myself节点
if (this == myself) continue;

// 2. 偏爱选择处于下线状态或疑似下线状态的节点
if (maxiterations > wanted*2 &&
!(this->flags & (CLUSTER_NODE_PFAIL|CLUSTER_NODE_FAIL)))
continue;

// 以下节点不能作为被选中的节点：
/*
1. 处于握手状态的节点
2. 带有NOADDR标识的节点
3. 因为不处理任何槽而断开连接的节点
*/
if (this->flags & (CLUSTER_NODE_HANDSHAKE|CLUSTER_NODE_NOADDR) ||
(this->link == NULL && this->numslots == 0))
{
freshnodes--; /* Tecnically not correct, but saves CPU. */
continue;
}

// 如果已经在gossip的消息中添加过了当前节点，则退出循环
for (j = 0; j < gossipcount; j++) {
if (memcmp(hdr->data.ping.gossip[j].nodename,this->name,
CLUSTER_NAMELEN) == 0) break;
}
// j 一定 == gossipcount
if (j != gossipcount) continue;

/* Add it */
// 这个节点满足条件，则将其添加到gossip消息中
freshnodes--;
// 指向添加该节点的那个空间
gossip = &(hdr->data.ping.gossip[gossipcount]);
// 添加名字
memcpy(gossip->nodename,this->name,CLUSTER_NAMELEN);
// 记录发送PING的时间
gossip->ping_sent = htonl(this->ping_sent);
// 接收到PING回复的时间
gossip->pong_received = htonl(this->pong_received);
// 设置该节点的IP和port
memcpy(gossip->ip,this->ip,sizeof(this->ip));
gossip->port = htons(this->port);
// 记录标识
gossip->flags = htons(this->flags);
gossip->notused1 = 0;
gossip->notused2 = 0;
// 已经添加到gossip消息的节点数加1
gossipcount++;
}

// 计算消息的总长度
totlen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
totlen += (sizeof(clusterMsgDataGossip)*gossipcount);
// 记录消息节点的数量到包头
hdr->count = htons(gossipcount);
// 记录消息节点的总长到包头
hdr->totlen = htonl(totlen);
// 发送消息
clusterSendMessage(link,buf,totlen);
zfree(buf);
}

freshnodes

int freshnodes = dictSize(server.cluster->nodes)-2;

freshnodes

freshnodes

wanted

wanted = floor(dictSize(server.cluster->nodes)/10);

wanted

wanted

gossip

Gossip

wanted

wanted

Redis Cluster

server.cluster_node_timeout*2

N

wanted

N/10

node_timeout

node_timeout/2

node_timeout

PING

PING

PING

PONG

node_timeout

Redis Cluster

server.cluster_node_timeout*2

node_timeout

N

wanted

N/10

8*N/10

80％

totlen

wanted

clusterBuildMessageHdr()

while

wanted

myself

myself

gossip

clusterSendMessage()

Gossip

3.3 分配槽位

Redis Cluster

0 ～ 16383

slot = CRC16(key)&16383

cluster addslots

6379

myself

{0..5461}

redis-cli -h 127.0.0.1 -p 6379 cluster addslots {0..5461}

3.3.1 槽位分配信息管理

6379

myself

{0..5461}

clusterNode

typedef struct clusterNode {
// 节点的槽位图
unsigned char slots[CLUSTER_SLOTS/8]; /* slots handled by this node */
// 当前节点复制槽的数量
int numslots;   /* Number of slots handled by this node */
} clusterNode;

6379

myself

clusterState

typedef struct clusterState {
// 导出槽数据到目标节点，该数组记录这些节点
clusterNode *migrating_slots_to[CLUSTER_SLOTS];
// 导入槽数据到目标节点，该数组记录这些节点
clusterNode *importing_slots_from[CLUSTER_SLOTS];
// 槽和负责槽节点的映射
clusterNode *slots[CLUSTER_SLOTS];
// 槽映射到键的跳跃表
zskiplist *slots_to_keys;
} clusterState;

migrating_slots_to

importing_slots_from

slots

slots_to_keys

CLUSTER GETKEYSINSLOT

3.3.2 分配槽位剖析

cluster addslots <slot> [slot ...]

cluster addslots

clusterCommand()

cluster

addslots

if ((!strcasecmp(c->argv[1]->ptr,"addslots") ||
!strcasecmp(c->argv[1]->ptr,"delslots")) && c->argc >= 3)
{
int j, slot;
unsigned char *slots = zmalloc(CLUSTER_SLOTS);
// 删除操作
int del = !strcasecmp(c->argv[1]->ptr,"delslots");

memset(slots,0,CLUSTER_SLOTS);
// 遍历所有指定的槽
for (j = 2; j < c->argc; j++) {
// 获取槽位的位置
if ((slot = getSlotOrReply(c,c->argv[j])) == -1) {
zfree(slots);
return;
}
// 如果是删除操作，但是槽没有指定负责的节点，回复错误信息
if (del && server.cluster->slots[slot] == NULL) {
addReplyErrorFormat(c,"Slot %d is already unassigned", slot);
zfree(slots);
return;
// 如果是添加操作，但是槽已经指定负责的节点，回复错误信息
} else if (!del && server.cluster->slots[slot]) {
addReplyErrorFormat(c,"Slot %d is already busy", slot);
zfree(slots);
return;
}
// 如果某个槽已经指定过多次了（在参数中指定了多次），那么回复错误信息
if (slots[slot]++ == 1) {
addReplyErrorFormat(c,"Slot %d specified multiple times",
(int)slot);
zfree(slots);
return;
}
}
// 上个循环保证了指定的槽的可以处理
for (j = 0; j < CLUSTER_SLOTS; j++) {
// 如果当前槽未指定
if (slots[j]) {
int retval;

// 如果这个槽被设置为导入状态，那么取消该状态
if (server.cluster->importing_slots_from[j])
server.cluster->importing_slots_from[j] = NULL;
// 执行删除或添加操作
retval = del ? clusterDelSlot(j) :
clusterAddSlot(myself,j);
serverAssertWithInfo(c,NULL,retval == C_OK);
}
}
zfree(slots);
// 更新集群状态和保存配置
clusterDoBeforeSleep(CLUSTER_TODO_UPDATE_STATE|CLUSTER_TODO_SAVE_CONFIG);
addReply(c,shared.ok);
}

clusterAddSlot()

myself

int clusterAddSlot(clusterNode *n, int slot) {
// 如果已经指定有节点，则返回C_ERR
if (server.cluster->slots[slot]) return C_ERR;
// 设置该槽被指定
clusterNodeSetSlotBit(n,slot);
// 设置负责该槽的节点n
server.cluster->slots[slot] = n;
return C_OK;
}

clusterNodeSetSlotBit()

myself

myself

int clusterNodeSetSlotBit(clusterNode *n, int slot) {
// 查看slot槽位是否被设置
int old = bitmapTestBit(n->slots,slot);
// 将slot槽位设置为1
bitmapSetBit(n->slots,slot);
// 如果之前没有被设置
if (!old) {
// 那么要更新n节点负责槽的个数
n->numslots++;
// 如果主节点是第一次指定槽，即使它没有从节点，也要设置MIGRATE_TO标识
// 当且仅当，至少有一个其他的主节点有从节点时，主节点就是有效的迁移目标
if (n->numslots == 1 && clusterMastersHaveSlaves())
// 设置节点迁移的标识，表示该节点可以迁移
n->flags |= CLUSTER_NODE_MIGRATE_TO;
}
return old;
}

3.3.3 广播节点的槽位信息

server.cluster->slots

clusterBuildMessageHdr()

clusterProcessPacket()

sender = clusterLookupNode(hdr->sender);
clusterNode *sender_master = NULL; /* Sender or its master if slave. */
int dirty_slots = 0; /* Sender claimed slots don't match my view? */

if (sender) {
// 如果sender是从节点，那么获取其主节点信息
// 如果sender是主节点，那么获取sender的信息
sender_master = nodeIsMaster(sender) ? sender : sender->slaveof;
if (sender_master) {
// sender发送的槽信息和主节点的槽信息是否匹配
dirty_slots = memcmp(sender_master->slots,
hdr->myslots,sizeof(hdr->myslots)) != 0;
}
}
// 1. 如果sender是主节点，但是槽信息出现不匹配现象
if (sender && nodeIsMaster(sender) && dirty_slots)
// 检查当前节点对sender的槽信息，并且进行更新
clusterUpdateSlotsConfigWith(sender,senderConfigEpoch,hdr->myslots);

sender

myself

myself

myself

sender

dirty_slots

clusterUpdateSlotsConfigWith()

myself

void clusterUpdateSlotsConfigWith(clusterNode *sender, uint64_t senderConfigEpoch, unsigned char *slots) {
int j;
clusterNode *curmaster, *newmaster = NULL;
uint16_t dirty_slots[CLUSTER_SLOTS];
int dirty_slots_count = 0;

// 如果当前节点是主节点，那么获取当前节点
// 如果当前节点是从节点，那么获取当前从节点所从属的主节点
curmaster = nodeIsMaster(myself) ? myself : myself->slaveof;
// 如果发送消息的节点就是本节点，则直接返回
if (sender == myself) {
serverLog(LL_WARNING,"Discarding UPDATE message about myself.");
return;
}
// 遍历所有槽
for (j = 0; j < CLUSTER_SLOTS; j++) {
// 如果当前槽已经被分配
if (bitmapTestBit(slots,j)) {
// 如果当前槽是sender负责的，那么跳过当前槽
if (server.cluster->slots[j] == sender) continue;
// 如果当前槽处于导入状态，它应该只能通过redis-trib 被手动修改，所以跳过该槽
if (server.cluster->importing_slots_from[j]) continue;
// 将槽重新绑定到新的节点，如果满足以下条件
/*
1. 该槽没有被指定或者新的节点声称它有一个更大的配置纪元
2. 当前没有导入该槽
*/
if (server.cluster->slots[j] == NULL ||
server.cluster->slots[j]->configEpoch < senderConfigEpoch)
{
// 如果当前槽被当前节点所负责，而且槽中有数据，表示该槽发生冲突
if (server.cluster->slots[j] == myself &&
countKeysInSlot(j) &&
sender != myself)
{
// 将发生冲突的槽记录到脏槽中
dirty_slots[dirty_slots_count] = j;
// 脏槽数加1
dirty_slots_count++;
}
// 如果当前槽属于当前节点的主节点，表示发生了故障转移
if (server.cluster->slots[j] == curmaster)
newmaster = sender;
// 删除当前被指定的槽
clusterDelSlot(j);
// 将槽分配给sender
clusterAddSlot(sender,j);
clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG|
CLUSTER_TODO_UPDATE_STATE|
CLUSTER_TODO_FSYNC_CONFIG);
}
}
}
// 如果至少一个槽被重新分配，从一个节点到另一个更大配置纪元的节点，那么可能发生了：
/*
1. 当前节点是一个不在处理任何槽的主节点，这是应该将当前节点设置为新主节点的从节点
2. 当前节点是一个从节点，并且当前节点的主节点不在处理任何槽，这是应该将当前节点设置为新主节点的从节点
*/
if (newmaster && curmaster->numslots == 0) {
serverLog(LL_WARNING,
"Configuration change detected. Reconfiguring myself "
"as a replica of %.40s", sender->name);
// 将sender设置为当前节点myself的主节点
clusterSetMaster(sender);
clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG|
CLUSTER_TODO_UPDATE_STATE|
CLUSTER_TODO_FSYNC_CONFIG);
} else if (dirty_slots_count) {
// 如果执行到这里，我们接收到一个删除当前我们负责槽的所有者的更新消息，但是我们仍然负责该槽，所以主节点不能被降级为从节点
// 为了保持键和槽的关系，需要从我们丢失的槽中将键删除
for (j = 0; j < dirty_slots_count; j++)
// 遍历所有的脏槽，删除槽中的键-
delKeysInSlot(dirty_slots[j]);
}
}

myself

sender

myself

redis-trib

myself

if

myself

dirty_slots

dirty_slots

myself

myself

myself

myself

sender

sender

myself

newmaster = sender

clusterSetMaster()

sender

myself

clusterAddSlot()

myself

sender

myself

16384