Redis是如何建立连接和处理命令的

今天来讲讲 Redis 的请求监听，通俗点说，就是Redis是如何监听客户端发出的set、get等命令的。

众所周知，Redis 是单进程单线程架构，虽然是单进程单线程，但是Redis的性能却毫不逊色，能轻松应对一般的高并发场景，那么Redis究竟是施了什么魔法呢？

其实 Redis 的原理和 Nginx 差不多，都利用了 IO 多路复用来提高处理能力，所谓多路复用，就是一个线程可以同时处理多个IO操作，当 某个 IO 操作 Ready 时，操作系统会主动通知进程。使用 IO 多路复用，我们可以使用 epoll、kqueue、select，API 都差不多。

与Redis 不同的是，Nginx 并不是单进程架构，而是采用了多进程来处理请求。Nginx跑起来后会先启动Maste进程，Master进程接着启动多个 Worker 进程，每个Worker 进程都会参与请求的监听和处理。这样可以充分发挥CPU的多核特性。

虽然 Redis 是单进程单线程，不能利用多核，但同样也避免了多进程的并发问题，也就没有了锁带来的开销。

Redis 入口是 server.c 中的 main（）方法，main（）中会调用 initServer（）初始化 Redis 服务。

Redis 为了进行事件监听，特地封装了一个 struct：aeEventLoop，定义在 ae.c 中。

为什么需要 aeEventLoop 呢？因为 Redis 针对于 epoll、evport、select 还有 kqueue 的 API做了封装，分别为 ae_epoll、ae_evport、ae_select 和 ae_kqueue。但是 Redis 在使用监听的时候，统一调用的都是 aeEventLoop 中的方法。ae.c 在预处理阶段使用了条件编译，这样便可以选择平台支持的 API 来使用多路复用。

#ifdef HAVE_EVPORT#include "ae_evport.c"#else    #ifdef HAVE_EPOLL    #include "ae_epoll.c"    #else        #ifdef HAVE_KQUEUE        #include "ae_kqueue.c"        #else        #include "ae_select.c"        #endif    #endif#endif

（1）创建内核事件队列

initServer（）第一件做的事情就是为当前服务创建了一个 aeEventLoop，后续的所有命令监听，都是基于 aeEventLoop 来做的。

server.el = aeCreateEventLoop(server.maxclients+CONFIG_FDSET_INCR);

aeCreateEventLoop（）内部会调用 aeApiCreate（），选择性的调用epoll、select 或者 kqueue 的 API，创建事件监听。这里拿 kqueue 来举例子，我们可以看下 ae_kqueue.c 中的实现。

static int aeApiCreate(aeEventLoop *eventLoop) {    aeApiState *state = zmalloc(sizeof(aeApiState));    ......    state->events = zmalloc(sizeof(struct kevent)*eventLoop->setsize);    state->kqfd = kqueue();    ......    eventLoop->apidata = state;    return 0;}

// 创建一个内核消息队列，返回队列描述符int  kqueue(void);

（2）创建套接字，监听端口

要想监听客户端的命令，首先要做的就是监听端口，initServer（）在创建好 aeEventLoop 后，就会开始执行端口监听操作：

listenToPort(server.port,server.ipfd,&server.ipfd_count) == C_ERR)

其中 server.ipfd 是个数组，用来存放端口监听后创建的描述符。为什么是个数组？因为Redis 可以同时 bind 多个 IP 地址，所以listenToPort（）内部会遍历配置文件中配置的多个IP地址，一次进行监听，并将创建的描述符存入 server.ipfd [] 中。

（3）注册监听到内核队列

监听好端口后，initServer（）会遍历刚刚创建好的套接字描述符 ipfd []，依次注册到事件监听中。

aeCreateFileEvent(server.el, server.ipfd[j], AE_READABLE, acceptTcpHandler,NULL)
我们进入aeCreateFileEvent（）中看看。

int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,        aeFileProc *proc, void *clientData){    ......    aeFileEvent *fe = &eventLoop->events[fd];
    if (aeApiAddEvent(eventLoop, fd, mask) == -1)        return AE_ERR;    ......    return AE_OK;}

static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {    aeApiState *state = eventLoop->apidata;    struct kevent ke;
    if (mask & AE_READABLE) {        EV_SET(&ke, fd, EVFILT_READ, EV_ADD, 0, 0, NULL);        if (kevent(state->kqfd, &ke, 1, NULL, 0, NULL) == -1) return -1;    }    if (mask & AE_WRITABLE) {        EV_SET(&ke, fd, EVFILT_WRITE, EV_ADD, 0, 0, NULL);        if (kevent(state->kqfd, &ke, 1, NULL, 0, NULL) == -1) return -1;    }    return 0;}

经过了上面的一系列初始化操作，Redis 就会正式进入事件监听的循环中，即 aeMain（）。

void aeMain(aeEventLoop *eventLoop) {    eventLoop->stop = 0;    while (!eventLoop->stop) {        if (eventLoop->beforesleep != NULL)            eventLoop->beforesleep(eventLoop);        aeProcessEvents(eventLoop, AE_ALL_EVENTS|AE_CALL_AFTER_SLEEP);    }}

aeProcessEvents（）中的核心逻辑就是两步：

（1）调用 aeApiPoll （）来等待接收事件。

numevents = aeApiPoll(eventLoop, tvp);

static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {    aeApiState *state = eventLoop->apidata;    int retval, numevents = 0;    ......    retval = kevent(state->kqfd, NULL, 0, state->events, eventLoop->setsize,                        NULL);    ......    numevents = retval;    ......    return numevents;}

// 用途：注册\反注册 监听事件，等待事件通知// kq，上面创建的消息队列描述符// changelist，需要注册的事件// changelist，changelist数组大小// eventlist，内核会把返回的事件放在该数组中// nevents，eventlist数组大小// timeout，等待内核返回事件的超时事件，NULL 即为无限等待int  kevent(int kq,          const struct kevent *changelist, int nchanges,         struct kevent *eventlist, int nevents,         const struct timespec *timeout);

（2）遍历返回的就绪事件数组，处理事件。

for (j = 0; j < numevents; j++) {    aeFileEvent *fe = &eventLoop->events[eventLoop->fired[j].fd];    int mask = eventLoop->fired[j].mask;    int fd = eventLoop->fired[j].fd;    int fired = 0; /* Number of events fired for current fd. */
    // 处理可读事件      if (!invert && fe->mask & mask & AE_READABLE) {        fe->rfileProc(eventLoop,fd,fe->clientData,mask);        fired++;    }
    // 处理可写事件    if (fe->mask & mask & AE_WRITABLE) {        if (!fired || fe->wfileProc != fe->rfileProc) {            fe->wfileProc(eventLoop,fd,fe->clientData,mask);            fired++;        }    }
    if (invert && fe->mask & mask & AE_READABLE) {        if (!fired || fe->wfileProc != fe->rfileProc) {            fe->rfileProc(eventLoop,fd,fe->clientData,mask);            fired++;        }    }
    processed++;}

记得 initServer（）时，Redis 调用 aeCreateFileEvent（）将监听端口后的套接字描述符注册到内核事件队列中吗？

aeCreateFileEvent(server.el, server.ipfd[j], AE_READABLE,            acceptTcpHandler,NULL) == AE_ERR)

在 Redis 服务端调用 accept（）与客户端建立连接后，会创建一个 client 对象，用来描述该连接。接着 Redis 会将accept（）返回的新连接的描述符再次调用aeCreateFileEvent（），注册到内核事件队列中，用来接收该连接后续的事件，说白了，就是监听后续该客户端的 set、get 等请求：

aeCreateFileEvent(server.el,fd,AE_READABLE,            readQueryFromClient, c) == AE_ERR

现在我们知道了用来处理客户端命令请求的函数是 readQueryFromClient（），由函数名就能看出它的意义：读取客户端的查询请求。

readQueryFromClient（）调用了 processInputBuffer（）从缓冲区读取并解析请求内容，之后 processInputBuffer（）便会调用 processCommand（）处理命令。

（1）查询命令

processCommand（）第一步要做的就是根据请求找到对应的命令。Redis 的所有命令都预先定义在 server.c 的 redisCommandTable [] 中：

struct redisCommand redisCommandTable[] = {    {"module",moduleCommand,-2,"as",0,NULL,0,0,0,0,0},    {"get",getCommand,2,"rF",0,NULL,1,1,1,0,0},    {"set",setCommand,-3,"wm",0,NULL,1,1,1,0,0},    {"setnx",setnxCommand,3,"wmF",0,NULL,1,1,1,0,0},    ......}

Redis 在启动后会调用 server.c 的 initServerConfig（）方法用来初始化配置，其中就调用了 populateCommandTable（）。populateCommandTable（）的作用就是遍历redisCommandTable []，将所有的命令以 name 作为 key，以自身 redisCommand 作为value，存入到一个字典中：server.commands。

processCommand（）调用了lookupCommand（）查询命令。

c->cmd = c->lastcmd = lookupCommand(c->argv[0]->ptr);
那么该去哪找对应的命令呢？当然就是上面所说这个字典中。

struct redisCommand *lookupCommand(sds name) {    return dictFetchValue(server.commands, name);}

（2）执行命令

找到请求对应的命令后，该干嘛？想都不用想，当然是去执行命令了。

if (c->flags & CLIENT_MULTI &&    c->cmd->proc != execCommand && c->cmd->proc != discardCommand &&    c->cmd->proc != multiCommand && c->cmd->proc != watchCommand){    queueMultiCommand(c);    addReply(c,shared.queued);} else {    call(c,CMD_CALL_FULL);    c->woff = server.master_repl_offset;    if (listLength(server.ready_keys))        handleClientsBlockedOnKeys();}

• 如果处于事务中，就将命令先放入队列中，不执行。

• 如果不在事务中，就直接调用call（）执行命令。

call（）中会调用命令对应的处理函数：

c->cmd->proc(c);

一步一步跟下来，对于网络请求这块，Redis 和 Nginx 确实太像了，都是用了IO多路复用。

Redis 监听命令主要就是下面几个步骤。

（5）当与客户端的TCP连接上有新事件时，调用对应的事件处理函数，从该连接上读取并解析请求，并执行请求对应的命令。