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

  1. 云栖社区>
  2. 博客>
  3. 正文

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

半栈工程师 2020-03-02 11:00:59 浏览315
展开阅读全文

一:前言

前几天写了如何调试Redis:《快速编译调试 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的多核特性。
想对 Nginx 多进程架构有更深了解的同学,可以看下我之前的一篇文章:《动手打造Nginx多进程架构》。
虽然 Redis 是单进程单线程,不能利用多核,但同样也避免了多进程的并发问题,也就没有了锁带来的开销。

三:源码探究

Redis 入口是 server.c 中的 main()方法,main()中会调用 initServer()初始化 Redis 服务。

3.1 初始化 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

因为笔者用的是 Mac ,Redis 选择了 Mac 平台支持的 kqueue 使用多路复用,所以后面源码分析的时候,我都会基于 kqueue 来讲解。
(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;
}

上面创建了一个 aeApiState,并且调用了 kqueue 的API:kqueue():

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

这里创建了一个内核事件队列,并将该队列的文件描述符存入了 state->kqfd 中。
(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;
}

上面主要是初始化了 aeFileEvent,存放在 eventLoop 中的events [] 内。并且调用 aeApiAddEvent(),将监听端口的描述符注册到(1)中创建的内核队列中。对于 kqueue来说,这里调用的是 kqueue 的 EV_SET():

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;
}

3.2 启动事件监听循环,监听请求

经过了上面的一系列初始化操作,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);
    }
}

这段代码看起来很简单,只要事件监听没有停止,Redis 就会一直循环调用aeProcessEvents()处理事件。
aeProcessEvents()中的核心逻辑就是两步:
(1)调用 aeApiPoll ()来等待接收事件。

numevents = aeApiPoll(eventLoop, tvp);

我们看下使用kqueue时,ae_kqueue.c 中的实现。

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;
}

这里调用的是 kevent()来接收事件。kevent()描述如下:

// 用途:注册\反注册 监听事件,等待事件通知
// 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++;
}

每次遍历,都会先从eventLoop 的 events[] 中,根据描述符取出对应的 aeFileEvent,接着调用 aeFileEvent 的 rfileProc()或者 wfileProc()来处理事件,那么 aeFileEvent 的 fileProc()时从哪来的呢?
记得 initServer()时,Redis 调用 aeCreateFileEvent()将监听端口后的套接字描述符注册到内核事件队列中吗?

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

此时传入的 acceptTcpHandler()就作为 rFileProc ,设置到此次创建的 aeFileEvent 之中 了。 acceptTcpHandler()的作用就是处理新客户端的连接请求,与新客户端建立 TCP 连接。
在 Redis 服务端调用 accept()与客户端建立连接后,会创建一个 client 对象,用来描述该连接。接着 Redis 会将accept()返回的新连接的描述符再次调用aeCreateFileEvent(),注册到内核事件队列中,用来接收该连接后续的事件,说白了,就是监听后续该客户端的 set、get 等请求:

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

到现在读者应该知道aeCreateFileEvent()的第四个参数时用来干嘛的了,就是监听到事件后用来处理事件的。可见用来处理客户端命令请求的函数是:readQueryFromClient()。

3.3 处理客户端命令

现在我们知道了用来处理客户端命令请求的函数是 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},
    ......
}

redisCommandTable [] 中定义了每个命令对应的处理函数,如set:setCommand, get:getCommand。
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);

如 set 命令对应的处理函数就是 t_string.c 中的 setCommand()。

四:总结

一步一步跟下来,对于网络请求这块,Redis 和 Nginx 确实太像了,都是用了IO多路复用。
Redis 监听命令主要就是下面几个步骤。
(1)创建套接字,监听端口,也就是监听新客户端的建立连接请求。
(2)创建内核事件队列,并注册上述的套接字描述符到队列中。
(3)开启循环,监听队列中的就绪事件。
(4)当端口有新事件时,调用 accept()与新客户端建立连接,并再次将新连接的描述符注册到内核事件队列中,监听该TCP连接上的事件。
(5)当与客户端的TCP连接上有新事件时,调用对应的事件处理函数,从该连接上读取并解析请求,并执行请求对应的命令。

网友评论

登录后评论
0/500
评论