深度解析单线程的 Redis 如何做到每秒数万 QPS 的超高处理能力( 五 ) _Redis

对于 get 命令来说，其对应的命令处理函数就是 getCommand 。也就是说当处理 GET 命令执行到 c->cmd->proc 的时候会进入到 getCommand 函数中来。
//file: src/t_string.cvoid getCommand(client *c) {getGenericCommand(c);}int getGenericCommand(client *c) {robj *o;if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.null[c->resp])) == NULL)return C_OK;...addReplyBulk(c,o);return C_OK;}getGenericCommand 方法会调用 lookupKeyReadOrReply 来从内存中查找对应的 key值。如果找不到，则直接返回 C_OK；如果找到了，调用 addReplyBulk 方法将值添加到输出缓冲区中。
//file: src/networking.cvoid addReplyBulk(client *c, robj *obj) {addReplyBulkLen(c,obj);addReply(c,obj);addReply(c,shared.crlf);}其主题是调用 addReply 来设置回复数据。在 addReply 方法中做了两件事情：

prepareClientToWrite 判断是否需要返回数据，并且将当前 client 添加到等待写返回数据队列中。
调用 _addReplyToBuffer 和 _addReplyObjectToList 方法将返回值写入到输出缓冲区中，等待写入 socekt

//file:src/networking.cvoid addReply(client *c, robj *obj) {if (prepareClientToWrite(c) != C_OK) return;if (sdsEncodedObject(obj)) {if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != C_OK)_addReplyStringToList(c,obj->ptr,sdslen(obj->ptr));} else {......}}

先来看 prepareClientToWrite 的详细实现，

//file: src/networking.cint prepareClientToWrite(client *c) {......if (!clientHasPendingReplies(c) && !(c->flags & CLIENT_PENDING_READ))clientInstallWriteHandler(c);}//file:src/networking.cvoid clientInstallWriteHandler(client *c) {c->flags |= CLIENT_PENDING_WRITE;listAddNodeHead(server.clients_pending_write,c);}

其中 server.clients_pending_write 就是我们说的任务队列，队列中的每一个元素都是有待写返回数据的 client 对象。在 prepareClientToWrite 函数中，把 client 添加到任务队列 server.clients_pending_write 里就算完事。
接下再来 _addReplyToBuffer ，该方法是向固定缓存中写，如果写不下的话就继续调用 _addReplyStringToList 往链表里写。简单起见，我们只看 _addReplyToBuffer 的代码。

//file:src/networking.cint _addReplyToBuffer(client *c, const char *s, size_t len) {......// 拷贝到 client 对象的 Response buffer 中memcpy(c->buf+c->bufpos,s,len);c->bufpos+=len;return C_OK;}

3.4 beforesleep 处理写任务队列回想在 aeMain 函数中，每次在进入 aeProcessEvents 前都需要先进行 beforesleep 处理。这个函数名字起的怪怪的，但实际上大有用处。

//file:src/ae.cvoid aeMain(aeEventLoop *eventLoop) {eventLoop->stop = 0;while (!eventLoop->stop) {// beforesleep 处理写任务队列并实际发送之if (eventLoop->beforesleep != NULL)eventLoop->beforesleep(eventLoop);aeProcessEvents(eventLoop, AE_ALL_EVENTS);}}

该函数处理了许多工作，其中一项便是遍历发送任务队列，并将 client 发送缓存区中的处理结果通过 write 发送到客户端手中。

文章插图

我们来看下 beforeSleep 的实际源码。
//file:src/server.cvoid beforeSleep(struct aeEventLoop *eventLoop) {......handleClientsWithPendingWrites();}

//file:src/networking.cint handleClientsWithPendingWrites(void) {listIter li;listNode *ln;int processed = listLength(server.clients_pending_write);//遍历写任务队列 server.clients_pending_writelistRewind(server.clients_pending_write,&li);while((ln = listNext(&li))) {client *c = listNodeValue(ln);c->flags &= ~CLIENT_PENDING_WRITE;listDelNode(server.clients_pending_write,ln);//实际将 client 中的结果数据发送出去writeToClient(c->fd,c,0)//如果一次发送不完则准备下一次发送if (clientHasPendingReplies(c)) {//注册一个写事件处理器 ， 等待 epoll_wait 发现可写后再处理aeCreateFileEvent(server.el, c->fd, ae_flags,sendReplyToClient, c);}......}}

在 handleClientsWithPendingWrites 中，遍历了发送任务队列 server.clients_pending_write ，并调用 writeToClient 进行实际的发送处理。
值得注意的是，发送 write 并不总是能一次性发送完的。假如要发送的结果太大，而系统为每个 socket 设置的发送缓存区又是有限的。
在这种情况下， clientHasPendingReplies 判断仍然有未发送完的数据的话，就需要注册一个写事件处理函数到 epoll 上。等待 epoll 发现该 socket 可写的时候再次调用 sendReplyToClient进行发送。