redis集群源码阅读 之 集群握手

Posted 2021-02-16 Seven_noon

集群节点的启动仍然是使用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连接，此时充当的是客户端。也就  是说每个节点即是服务端又是客户端。

　　

　　时间事件的实现