集群节点的启动仍然是使用redis-server命令，但需要使用集群模式启动。启动完之后各个节点分别在各自的集群内，可以通过cluster meet命令将两个节点加入到同一个集群。集群相关的命令通过[email protected]这个api现实。下面主要通过源码分析来看看A节点向B节点发送cluster meet命令的过程。
- 处理cluster meet命令的整个流程
if (!strcasecmp(c->argv[1]->ptr,"meet") && c->argc == 4) {
/* CLUSTER MEET <ip> <port> */
// 将给定地址的节点添加到当前节点所处的集群里面
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);
}
}
可以看到检查ip:port之后就是进程cluster handshake。
- 节点之间握手
B节点首先根据提供的ip创建一个带有REDIS_NODE_HANDSHAKE|REDIS_NODE_MEET标志的集群节点，并把A节点加入到集群中。并给A节点赋予一个随机的名字。
n = createClusterNode(NULL,REDIS_NODE_HANDSHAKE|REDIS_NODE_MEET); memcpy(n->ip,norm_ip,sizeof(n->ip)); n->port = port; // 将节点添加到集群当中 clusterAddNode(n);
当以集群模式启动redis-server时，在时间事件循环serverCron中，以每秒10次的频率会执行clusterCron。 在clusterCron中会对集群的节点（cluster->nodes）做一些检查和处理。
刚才创建的A节点，还是新建的状态，node->link还是NULL。 此时会向A节点创建一个tcp连接，用于后面节点之间的通信。
if (node->link == NULL) {
int fd;
mstime_t old_ping_sent;
clusterLink *link;
fd = anetTcpNonBlockBindConnect(server.neterr, node->ip,
node->port+REDIS_CLUSTER_PORT_INCR,
server.bindaddr_count ? server.bindaddr[0] : NULL);
if (fd == -1) {
redisLog(REDIS_DEBUG, "Unable to connect to "
"Cluster Node [%s]:%d -> %s", node->ip,
node->port+REDIS_CLUSTER_PORT_INCR,
server.neterr);
continue;
}
link = createClusterLink(node);
link->fd = fd;
node->link = link;
aeCreateFileEvent(server.el,link->fd,AE_READABLE,
clusterReadHandler,link);
/* Queue a PING in the new connection ASAP: this is crucial
* to avoid false positives in failure detection.
*
* If the node is flagged as MEET, we send a MEET message instead
* of a PING one, to force the receiver to add us in its node
* table. */
// 向新连接的节点发送 PING 命令，防止节点被识进入下线
// 如果节点被标记为 MEET ，那么发送 MEET 命令，否则发送 PING 命令
old_ping_sent = node->ping_sent;
clusterSendPing(link, node->flags & REDIS_NODE_MEET ?
CLUSTERMSG_TYPE_MEET : CLUSTERMSG_TYPE_PING);}
其中ClusterLink用于处理与A节点之间的读写（有读写缓存）操作。创建完tcp之后会立即发送一条CLUSTERMSG_TYPE_MEET类型Ping命令给节点A。
A节点收到B节点发来的Ping消息（处理消息的API为clusterProcessPacket）
if (type == CLUSTERMSG_TYPE_PING || type == CLUSTERMSG_TYPE_MEET) {
redisLog(REDIS_DEBUG,"Ping packet received: %p", (void*)link->node);
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);
}
会将B节点加入到cluster->nodes中，并设置标志位REDIS_NODE_HANDSHAKE。然后回复B节点一个带有A节点信息的PONG命令。B节点收到A节点发来的Pong命令，就会更新A节点的信息。
同时，在A节点的clusterCron，也会处理新创建的节点B（作为客户端连上B节点的cfd，发送ping命令），至此握手完成。
- 一些小细节
节点的cfd是以port+REDIS_CLUSTER_PORT_INCR为端口创建的socket描述符，充当的是服务器。 同时，集群中的其他节点会向目标节点的port+REDIS_CLUSTER_PORT_INCR端口建立一个tcp连接，此时充当的是客户端。也就 是说每个节点即是服务端又是客户端。
时间事件的实现
原文地址：https://www.cnblogs.com/coderht/p/10705959.html