因为针对 READ 事件的 clusterReadHandler 处理器主要工作就是解析 cluster bus 上接收的数据并进行消息分包，然后对消息进行处理，而对于消息的分包首先需要了解一下消息结构，Redis Cluster 节点之间通信的消息结构定义如下:
typedef struct {
char sig[4]; /* Siganture "RCmb" (Redis Cluster message bus). */
uint32_t totlen; /* Total length of this message */
uint16_t ver; /* Protocol version, currently set to 0. */
uint16_t notused0; /* 2 bytes not used. */
/* Message type，如：PING, PONG，定义参考宏定义 CLUSTERMSG_TYPE_* */
uint16_t type;
uint16_t count; /* Only used for some kind of messages. */
uint64_t currentEpoch; /* The epoch accordingly to the sending node. */
uint64_t configEpoch; /* The config epoch if it‘s a master, or the last
epoch advertised by its master if it is a
slave. */
uint64_t offset; /* Master replication offset if node is a master or
processed replication offset if node is a slave. */
// 节点发送方，为 NodeID 表示，如: 123ed65d59ff22370f2f09546f410d31207789f6
char sender[REDIS_CLUSTER_NAMELEN]; /* Name of the sender node */
// 本节点维护的 slots bits
unsigned char myslots[REDIS_CLUSTER_SLOTS/8];
// 如果本节点为 slave 节点，则 slaveof 记录对应的 master 节点ID
char slaveof[REDIS_CLUSTER_NAMELEN];
char notused1[32]; /* 32 bytes reserved for future usage. */
uint16_t port; /* Sender TCP base port */
uint16_t flags; /* Sender node flags */
// cluster 状态， 如：REDIS_CLUSTER_OK, REDIS_CLUSTER_FAIL ...
unsigned char state; /* Cluster state from the POV of the sender */
unsigned char mflags[3]; /* Message flags: CLUSTERMSG_FLAG[012]_... */
// 指向不同 消息类型 的消息体
// 参考 clusterMsgData 结构的说明
union clusterMsgData data;
} clusterMsg;
从上面结构可以看到消息分包，主要解析前 8 个字节，分别为:
- char sig[4]; // 消息签名，对于 cluster 消息，固定为字符序列 RCmb
- uint32_t totlen; // 消息总长度
其他结构成员都是在处理消息时使用的，后续讲解消息处理流程时进行分析。
/* Read data. Try to read the first field of the header first to check the
* full length of the packet. When a whole packet is in memory this function
* will call the function to process the packet. And so forth. */
void clusterReadHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
char buf[sizeof(clusterMsg)];
ssize_t nread;
clusterMsg *hdr;
clusterLink *link = (clusterLink*) privdata;
int readlen, rcvbuflen;
REDIS_NOTUSED(el);
REDIS_NOTUSED(mask);
while(1) { /* Read as long as there is data to read. */
rcvbuflen = sdslen(link->rcvbuf);
if (rcvbuflen < 8) {
/* First, obtain the first 8 bytes to get the full message
* length. */
readlen = 8 - rcvbuflen;
} else {
// 已经知道了本条消息的长度
// 本块代码主要计算剩余还需读入的字节数(readlen)才是完整的消息
/* Finally read the full message. */
hdr = (clusterMsg*) link->rcvbuf;
if (rcvbuflen == 8) {
/* Perform some sanity check on the message signature
* and length. */
if (memcmp(hdr->sig,"RCmb",4) != 0 ||
ntohl(hdr->totlen) < CLUSTERMSG_MIN_LEN)
{
redisLog(REDIS_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);
}
// 读入本条消息记录的剩余 readlen 个字节的数据
// 因为这里的 fd 是非阻塞的，所以需要判断 EAGAIN
nread = read(fd,buf,readlen);
if (nread == -1 && errno == EAGAIN) return; /* No more data ready. */
if (nread <= 0) {
/* I/O error... */
redisLog(REDIS_DEBUG,"I/O error reading from node link: %s",
(nread == 0) ? "connection closed" : strerror(errno));
handleLinkIOError(link);
return;
} else {
/* Read data and recast the pointer to the new buffer. */
link->rcvbuf = sdscatlen(link->rcvbuf,buf,nread);
hdr = (clusterMsg*) link->rcvbuf;
rcvbuflen += nread;
}
/* Total length obtained? Process this packet. */
if (rcvbuflen >= 8 && rcvbuflen == ntohl(hdr->totlen)) {
// 表明 link 上的 rcvbuf 已经是一个完整的 cluster 消息
// 下面开始处理此消息
if (clusterProcessPacket(link)) {
sdsfree(link->rcvbuf);
link->rcvbuf = sdsempty();
} else {
return; /* Link no longer valid. */
}
}
}
}本文出自 “安静的疯子” 博客，请务必保留此出处http://quietmadman.blog.51cto.com/3269500/1558889
【原创】 Redis Cluster 的实现 - cluster 消息的接收和分包
原文地址：http://quietmadman.blog.51cto.com/3269500/1558889
