0. 写在前面

Version Redis2.2.2

Redis中可以支持主从结构，本文主要从master和slave的心跳机制出发(PING)，分析redis的命令行交互。

在Redis中，server为每个连接建立一个redisClient数据对象，来描述对应的连接。其中，redisClient为命令交互设置了缓冲区。querybuf用于存储客户端送过来的命令，buf和reply是用于应答的缓冲。querybuf是在文件事件readQueryFromClient中被填充，每次填充的最大字节数默认为1024B。而应答缓冲区是由addReply()函数填充，并由文件事件sendReplyToClient中发送给客户端。具体数据流如图1所示。MasterPorcess与SlaveProcess进行命令交互。其中，蓝色矩形框代表函数，白色矩形框代表数据，曲线描述数据流，折线描述数据间的从属关系。

图1. Master&Slave交互的数据流(蓝色矩形框代表函数，白色矩形框代表数据，曲线描述数据流，折线描述数据间的从属关系)

1. 相关数据结构

typedef struct redisClient {int fd;                   //connect fd...sds querybuf;            //命令缓冲区,由readQueryFromClient()事件进行填充(sds equals to char*)int argc;              //for command；记录参数个数robj **argv;         //for command；记录命令行参数int reqtype;            //命令解析协议：INLINE or MULTIBULK...time_t lastinteraction; /* 最近交互时间 */...list *reply;           //Replay object list/* Response buffer */char buf[REDIS_REPLY_CHUNK_BYTES]; //Reply buffer,由addReply()函数进行填充int bufpos;             //记录buf已填充的长度int sentlen;           //Replay阶段，记录当前buf已发送了多少字节
} redisClient;struct redisServer {...list *clients;dict *commands;             /* Command table hahs table */...list *slaves, *monitors;    //Master ： slave链表char neterr[ANET_ERR_LEN];aeEventLoop *el;         //Event listint cronloops;              //ServerCorn 执行次数...redisClient *master;    //Slave ：记录 master 的连接信息的clientint replstate;          //Slave ：当前的状态...
};struct redisCommand readonlyCommandTable[] = {...{"sync",syncCommand,1,0,NULL,0,0,0},...{"ping",pingCommand,1,0,NULL,0,0,0},...
}

2. query的读取和命令的解析

从图1可以看出，命令交互数据query的读取是在文件事件readQueryFromClient中填充到c->querybuf中。之后，querybuf由函数processInputBuffer进行命令的解析。命令的解析过程如图2所示。在函数processInputBuffer中，将缓存与querybuf中的所有命令(命令间按\n\r分隔)进行解析。之后，查询命令hashtabe查找相关命令函数。最后调用相应命令hander执行命令。

图2.querybuf的解析

具体代码分析如下：

void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {redisClient *c = (redisClient*) privdata;char buf[REDIS_IOBUF_LEN];int nread;REDIS_NOTUSED(el);REDIS_NOTUSED(mask);nread = read(fd, buf, REDIS_IOBUF_LEN);...check...if (nread) {c->querybuf = sdscatlen(c->querybuf,buf,nread);c->lastinteraction = time(NULL);//更新时间戳} else {return;}processInputBuffer(c);//处理client传输过来的数据
}void processInputBuffer(redisClient *c) {/* 执行querybub中的所有命令*/while(sdslen(c->querybuf)) {...check.../*判定命令的解析协议 */if (!c->reqtype) {if (c->querybuf[0] == '*') {c->reqtype = REDIS_REQ_MULTIBULK;} else {c->reqtype = REDIS_REQ_INLINE;//按行解析}}if (c->reqtype == REDIS_REQ_INLINE) {/*processInlineBuffer: 1. 取出c->querybuf起始端到\r\n位置的字符串，更新c->querybuf2. 将取出的字符串按照“ ”空格进行分段解析，得到命令及其参数格式为： argc，*argv[],其中argv[0]为命令，argv[1~argc-1]为参数*/if (processInlineBuffer(c) != REDIS_OK) break;} else if (c->reqtype == REDIS_REQ_MULTIBULK) {...}/* Multibulk processing could see a <= 0 length. */if (c->argc == 0) {resetClient(c);} else {/* Only reset the client when the command was executed. */if (processCommand(c) == REDIS_OK)   //执行命令resetClient(c);}}
}/* If this function gets called we already read a whole* command, argments are in the client argv/argc fields.* processCommand() execute the command or prepare the* server for a bulk read from the client.*/
int processCommand(redisClient *c) {struct redisCommand *cmd;.../* Now lookup the command and check ASAP about trivial error conditions* such wrong arity, bad command name and so forth. */cmd = lookupCommand(c->argv[0]->ptr);...check.../* Exec the command */if (c->flags & REDIS_MULTI &&cmd->proc != execCommand && cmd->proc != discardCommand &&cmd->proc != multiCommand && cmd->proc != watchCommand){queueMultiCommand(c,cmd);addReply(c,shared.queued);} else {if (server.vm_enabled && server.vm_max_threads > 0 &&blockClientOnSwappedKeys(c,cmd)) return REDIS_ERR;call(c,cmd);    //执行命令}return REDIS_OK;
}/* Call() is the core of Redis execution of a command */
void call(redisClient *c, struct redisCommand *cmd) {long long dirty;dirty = server.dirty;cmd->proc(c);     //执行命令dirty = server.dirty-dirty;if (server.appendonly && dirty)feedAppendOnlyFile(cmd,c->db->id,c->argv,c->argc);if ((dirty || cmd->flags & REDIS_CMD_FORCE_REPLICATION) &&listLength(server.slaves))replicationFeedSlaves(server.slaves,c->db->id,c->argv,c->argc);if (listLength(server.monitors))replicationFeedMonitors(server.monitors,c->db->id,c->argv,c->argc);server.stat_numcommands++;
}

3. 具体命令的执行（ping命令）

其中，addReply将相关命令执行结果放入client的reply缓冲区中。reply缓冲区的发送时机是在事件sendReplyToClient中进行。

#define REDIS_STRING 0
shared.pong = createObject(REDIS_STRING,sdsnew("+PONG\r\n"));
//{"ping",pingCommand,1,0,NULL,0,0,0}
void pingCommand(redisClient *c) {addReply(c,shared.pong); //ping的回复是pong，打乒乓，呵呵
}//将命令执行的返回结构写入c->buf 或者 c->reply
void addReply(redisClient *c, robj *obj) {if (_installWriteEvent(c) != REDIS_OK) return;//创建event sendReplyToClientredisAssert(!server.vm_enabled || obj->storage == REDIS_VM_MEMORY);/* This is an important place where we can avoid copy-on-write* when there is a saving child running, avoiding touching the* refcount field of the object if it's not needed.** If the encoding is RAW and there is room in the static buffer* we'll be able to send the object to the client without* messing with its page. */if (obj->encoding == REDIS_ENCODING_RAW) {if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != REDIS_OK)_addReplyObjectToList(c,obj);} else {/* FIXME: convert the long into string and use _addReplyToBuffer()* instead of calling getDecodedObject. As this place in the* code is too performance critical. */obj = getDecodedObject(obj);if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != REDIS_OK)_addReplyObjectToList(c,obj);decrRefCount(obj);}
}

4. reply缓冲区数据的发送

将c->buf 和 c->reply中的数据发送到客户端(slave or master)。在每次文件事件中发送所有的reply缓冲区中的数据。

void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) {redisClient *c = privdata;int nwritten = 0, totwritten = 0, objlen;robj *o;REDIS_NOTUSED(el);REDIS_NOTUSED(mask);while(c->bufpos > 0 || listLength(c->reply)) {if (c->bufpos > 0) {//发送c->buf中的数据if (c->flags & REDIS_MASTER) {/* Don't reply to a master */nwritten = c->bufpos - c->sentlen;} else {nwritten = write(fd,c->buf+c->sentlen,c->bufpos-c->sentlen);if (nwritten <= 0) break;}c->sentlen += nwritten;totwritten += nwritten;/* If the buffer was sent, set bufpos to zero to continue with* the remainder of the reply. */if (c->sentlen == c->bufpos) {c->bufpos = 0;c->sentlen = 0;}} else {//发送c->reply中的数据o = listNodeValue(listFirst(c->reply));objlen = sdslen(o->ptr);if (objlen == 0) {listDelNode(c->reply,listFirst(c->reply));continue;}if (c->flags & REDIS_MASTER) {/* Don't reply to a master */nwritten = objlen - c->sentlen;} else {nwritten = write(fd, ((char*)o->ptr)+c->sentlen,objlen-c->sentlen);if (nwritten <= 0) break;}c->sentlen += nwritten;totwritten += nwritten;/* If we fully sent the object on head go to the next one */if (c->sentlen == objlen) {listDelNode(c->reply,listFirst(c->reply));c->sentlen = 0;}}/* Note that we avoid to send more thank REDIS_MAX_WRITE_PER_EVENT* bytes, in a single threaded server it's a good idea to serve* other clients as well, even if a very large request comes from* super fast link that is always able to accept data (in real world* scenario think about 'KEYS *' against the loopback interfae) */if (totwritten > REDIS_MAX_WRITE_PER_EVENT) break;}...check...if (totwritten > 0) c->lastinteraction = time(NULL);//??Why delete file event of write ? ?if (listLength(c->reply) == 0) {c->sentlen = 0;aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE);/* Close connection after entire reply has been sent. */if (c->flags & REDIS_CLOSE_AFTER_REPLY) freeClient(c);}
}

5. 总结

命令行交互过程中，1.为每个连接有相应的数据进行描述（redisClient），这样便于连接的管理。2.命令行交互中，引入命令缓冲区querybuf，这样可以延时处理命令，这在事件轮询机制中，是至关重要的。

原文链接 http://blog.csdn.net/ordeder/article/details/16105345