redis集群实现（二）集群添加节点

在redis-3.0.0里，集群添加节点是通过客户端运行cluster meet命令来实现的，命令格式是cluster meet <ip> <port>，如果客户端向A节点发送这条命令，ip和port分别是B节点的ip和port，就会把ip:port的机器添加进入执行命令的节点所在的集群里。
具体的流程如下：
1.首先客户端向A节点发送cluster meet <ip> <port>命令。
2.A节点在本地为B节点创建对应的数据结构，然后向B节点发送meet命令。
3.B节点在本地为A节点创建相应的数据结构，并向A节点发送PONG消息，表示收到A节点的消息。
4.A节点收到PONG以后，向B节点返回PING消息。
5.A节点通过Gossip协议向集群中的其他节点传播，一段时间以后，集群中所有的节点都会知道B。
如图：



理解了原理了以后，我们就来看一看redis-3.0.0里对于cluster meet命令以及后续的过程的代码实现，以便于更加深入的理解redis。
redis的meet命令的定义是clusterCommand函数，我们看看clusterCommand函数对于meet的实现。

void clusterCommand(redisClient *c) {
// 不能在非集群模式下使用该命令
if (server.cluster_enabled == 0) {
addReplyError(c,"This instance has cluster support disabled");
return;
}
//匹配命令，如果是meet就进入下边的代码
if (!strcasecmp(c->argv[1]->ptr,"meet") && c->argc == 4) {
// 将给定地址的节点添加到当前节点所处的集群里面

long long port;

// 检查 port 参数的合法性
if (getLongLongFromObject(c->argv[3], &port) != REDIS_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);
} else {
// 连接成功
addReply(c,shared.ok);
}

}
……………………………………………….

主要的处理逻辑函数是clusterStartHandshake，这个函数向ip:port进行握手，成功时返回1，我们看看clusterStartHandshake函数的代码实现吧

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

// ip 合法性检查
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;
}

// port 合法性检查
if (port <= 0 || port > (65535-REDIS_CLUSTER_PORT_INCR)) {
errno = EINVAL;
return 0;
}

if (sa.ss_family == AF_INET)
inet_ntop(AF_INET,
(void*)&(((struct sockaddr_in *)&sa)->sin_addr),
norm_ip,REDIS_IP_STR_LEN);
else
inet_ntop(AF_INET6,
(void*)&(((struct sockaddr_in6 *)&sa)->sin6_addr),
norm_ip,REDIS_IP_STR_LEN);

// 检查节点是否已经发送握手请求，如果是的话，那么直接返回，防止出现重复握手
if (clusterHandshakeInProgress(norm_ip,port)) {
errno = EAGAIN;
return 0;
}

// 对给定地址的节点设置一个随机名字
// 当 HANDSHAKE 完成时，当前节点会取得给定地址节点的真正名字
// 创建一个集群节点，flag设置为MEET，发出meet命令
n = createClusterNode(NULL,REDIS_NODE_HANDSHAKE|REDIS_NODE_MEET);
memcpy(n->ip,norm_ip,sizeof(n->ip));
n->port = port;

// 将节点添加到集群当中
clusterAddNode(n);

return 1;
}

我们继续看createClusterNode函数和clusterAddNode函数。

createClusterNode函数创建了一个ClusterNode结构体，并且设置状态为handshake和meet
/*
* 函数会返回一个被创建的节点，但是并没有把它加入到当前节点的哈希表里
*/
clusterNode *createClusterNode(char *nodename, int flags) {
clusterNode *node = zmalloc(sizeof(*node));

// 如果没有指定节点名字，就采用随机的，获取返回信息以后就会设置真正的节点名字
if (nodename)
memcpy(node->name, nodename, REDIS_CLUSTER_NAMELEN);
else
getRandomHexChars(node->name, REDIS_CLUSTER_NAMELEN);

// 初始化属性
node->ctime = mstime();
node->configEpoch = 0;
//这里设置flags为传入的值
node->flags = flags;
memset(node->slots,0,sizeof(node->slots));
node->numslots = 0;
node->numslaves = 0;
node->slaves = NULL;
node->slaveof = NULL;
node->ping_sent = node->pong_received = 0;
node->fail_time = 0;
node->link = NULL;
memset(node->ip,0,sizeof(node->ip));
node->port = 0;
node->fail_reports = listCreate();
node->voted_time = 0;
node->repl_offset_time = 0;
node->repl_offset = 0;
listSetFreeMethod(node->fail_reports,zfree);

return node;
}

然后是clusterAddNode函数，这个函数把一个刚刚创建好的节点加入到当前节点哈希表里边，代码非常简单。

// 将给定 node 添加到节点表里面
int clusterAddNode(clusterNode *node) {
int retval;
// 将 node 添加到当前节点的 nodes表中
// 这样接下来当前节点就会创建连向 node的节点
retval = dictAdd(server.cluster->nodes,
sdsnewlen(node->name,REDIS_CLUSTER_NAMELEN), node);
return (retval == DICT_OK) ? REDIS_OK : REDIS_ERR;
}

看到这里，你可能会问，怎么没有看到发送MEET信息的代码？其实MEET信息是在serverCron函数里边发送的，serverCron函数是一个周期性执行的函数，一般是每秒调用10次，就是每100ms调用一次。serverCron函数的功能是清除一些过期的key-value和统计信息，复制等一些操作。在serverCron函数里有以下代码：

// 如果服务器运行在集群模式下，那么执行集群操作
run_with_period(100) {
if (server.cluster_enabled) clusterCron();
}

这里就是每100ms会执行clusterCron函数，clusterCron函数执行集群的定期检查工作，在clusterCron函数里执行了发送MEET消息的工作，具体实现如下：

void clusterCron(void) {
………………
//如果当前节点没有创建连接
if (node->link == NULL) {
……………….
//如果当前节点有MEET标记就发送MEET消息
old_ping_sent = node->ping_sent;
clusterSendPing(link, node->flags & REDIS_NODE_MEET ?
CLUSTERMSG_TYPE_MEET : CLUSTERMSG_TYPE_PING);
……………………
}

自此，我们就把MEET消息发送出去了~~然后，我们就看看另一个节点是怎么接收到MEET消息的吧。
又回到clusterCron函数，在clusterCron函数里，为没有创建连接的节点结构体设置对应的消息处理函数，代码如下

if (node->link == NULL) {
……………………
aeCreateFileEvent(server.el,link->fd,AE_READABLE,
clusterReadHandler,link);
……………………
clusterReadHandler函数调用了clusterProcessPacket函数处理消息包，我们来看clusterProcessPacket函数：
………………………………………..
if (type == CLUSTERMSG_TYPE_PING || type == CLUSTERMSG_TYPE_MEET) {
redisLog(REDIS_DEBUG,"Ping packet received: %p", (void*)link->node);

/*
* 如果当前节点是第一次遇见这个节点，并且对方发来的是 MEET信息，
* 那么将这个节点添加到集群的节点列表里面。
* 节点目前的 flag 、 slaveof等属性的值都是未设置的，
* 等当前节点向对方发送 PING 命令之后，
* 这些信息可以从对方回复的 PONG信息中取得。
*/
if (!sender && type == CLUSTERMSG_TYPE_MEET) {
clusterNode *node;

// 创建 HANDSHAKE 状态的新节点
node = createClusterNode(NULL,REDIS_NODE_HANDSHAKE);

// 设置 IP 和端口
nodeIp2String(node->ip,link);
node->port = ntohs(hdr->port);

// 将新节点添加到集群
clusterAddNode(node);

clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG);
}

/* Get info from the gossip section */
// 分析并取出消息中的 gossip 节点信息
clusterProcessGossipSection(hdr,link);

/* Anyway reply with a PONG */
// 向目标节点返回一个 PONG
clusterSendPing(link,CLUSTERMSG_TYPE_PONG);
}
………………………………………..

节点发送了PONG，我们就该接收PONG，同样是在clusterProcessPacket函数，接受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 &= ~REDIS_NODE_PFAIL;

clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG|
CLUSTER_TODO_UPDATE_STATE);
} else if (nodeFailed(link->node)) {
// 看是否可以撤销 FAIL
clearNodeFailureIfNeeded(link->node);
}
}
………………………………………

接收到PONG消息以后，撤销了节点的Fail状态，以后就会在ServerCron函数里周期性向新加入节点发送PING，然后新加入的节点返回PONG，这方面的代码会在后续的专题写，自此双方的信息就都了解清楚了。