Redis Cluster 的实现 - cluster 消息的接收和分包

人气旺

因为针对 READ 事件的 clusterReadHandler 处理器主要工作就是解析 cluster bus 上接收的数据并进行消息分包，然后对消息进行处理，而对于消息的分包首先需要了解一下消息结构，Redis Cluster 节点之间通信的消息结构定义如下:
  

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38

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;  // 消息总长度
  
其他结构成员都是在处理消息时使用的，后续讲解消息处理流程时进行分析。

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68

/* 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. */             }         }     } }