从一条Redis命令是如何执行了解事件驱动

水深无声 2023-09-26 17:20 22阅读 0赞

> 大家在使用 Redis 的时候有没有想过一条命令是如何执行的呢？下面我带领大家了解一下 Redis 命令执行的原理。

## Redis 命令执行流程 ##

### 客户端链接 ###

当 `Redis server` 接收到客户端的连接请求时，就会使⽤注册好的 `acceptTcpHandler` 函数进⾏处理。`acceptTcpHandler` 函数会接受客户端连接，并创建已连接套接字 `cfd`。然后把刚刚创建好的已连接套接字 `cfd` 作为参数调用 `acceptCommonHandler` 函数。

`acceptCommonHandler` 函数会调用 `createClient` 函数来创建客户端。在 `createClient` 函数中调用 `aeCreateFileEvent` 来添加可读事件的回调函数。具体调用如下：

aeCreateFileEvent(server.el,fd,AE_READABLE,
                readQueryFromClient, c)
    复制代码

可以看到监听的事件类型是 `AE_READABLE`,注册的回调函数是 `readQueryFromClient`,现在我们了解到当有读事件时会调用 `readQueryFromClient` 来处理。

### 读事件处理 ###

当服务器接收到客户端的命令时会触发可读事件，调用 `readQueryFromClient` 函数。`readQueryFromClient` 函数首先会将命令写入输入缓冲区，然后从输入缓冲区中解析发送过来的命令，调用 `processCommand` 函数来执行命令，最后将返回给客户端的内容写入输出缓冲区。

### 写事件处理 ###

`Redis` 事件驱动框架每次循环进入事件处理函数之前，也就是在调用 `aeProcessEvents` 函数前，会调用 `beforeSleep` 函数进行一些任务处理，其中会调用 `handleClientsWithPendingWrites` 函数，它会将 `Redis` 的输出缓冲区的数据写回客户端。我们看一下事件循环的主函数 `aeMain`：

void aeMain(aeEventLoop *eventLoop) {
        eventLoop->stop = 0;
        while (!eventLoop->stop) {
            if (eventLoop->beforesleep != NULL)
                // 调用 beforesleep 函数
                eventLoop->beforesleep(eventLoop);
            aeProcessEvents(eventLoop, AE_ALL_EVENTS|AE_CALL_AFTER_SLEEP);
        }
    }
    复制代码

`handleClientsWithPendingWrites` 会遍历每个代写回数据的客户端，并调用 `writeToClient` 写回数据。如果还有待写回数据，则会创建可写事件监听设置回调函数`sendReplyToClient`, `sendReplyToClient` 这个函数也会调用 `writeToClient` 写回数据。我们来看一下 `handleClientsWithPendingWrites` 函数的基本流程：

int handleClientsWithPendingWrites(void) {
        listIter li;
        listNode *ln;
        int processed = listLength(server.clients_pending_write);
        // 获取待写回的客户端列表
        listRewind(server.clients_pending_write,&li);
        //遍历所有待写会客户端
        while((ln = listNext(&li))) {
            client *c = listNodeValue(ln);
            c->flags &= ~CLIENT_PENDING_WRITE;
            listDelNode(server.clients_pending_write,ln);
    
            /* If a client is protected, don't do anything,
             * that may trigger write error or recreate handler. */
            if (c->flags & CLIENT_PROTECTED) continue;
    
            /* Try to write buffers to the client socket. */
            // 将客户端输出缓冲区的数据写回
            if (writeToClient(c->fd,c,0) == C_ERR) continue;
    
            /* If after the synchronous writes above we still have data to
             * output to the client, we need to install the writable handler. */
            // 如果还有待写回数据
            if (clientHasPendingReplies(c)) {
                int ae_flags = AE_WRITABLE;
                /* For the fsync=always policy, we want that a given FD is never
                 * served for reading and writing in the same event loop iteration,
                 * so that in the middle of receiving the query, and serving it
                 * to the client, we'll call beforeSleep() that will do the
                 * actual fsync of AOF to disk. AE_BARRIER ensures that. */
                if (server.aof_state == AOF_ON &&
                    server.aof_fsync == AOF_FSYNC_ALWAYS)
                {
                    ae_flags |= AE_BARRIER;
                }
                // 创建可写事件的监听并设置回调函数
                if (aeCreateFileEvent(server.el, c->fd, ae_flags,
                    sendReplyToClient, c) == AE_ERR)
                {
                        freeClientAsync(c);
                }
            }
        }
        return processed;
    }
    复制代码

## 事件驱动框架 ##

`Redis` 事件驱动框架有两大类事件：`IO` 事件和时间事件。下面我们来学习一下他们相应的处理机制。

### aeEventLoop 结构体与初始化 ###

/* State of an event based program */
    typedef struct aeEventLoop {
        // 当前最大的 fd
        int maxfd;   /* highest file descriptor currently registered */
        // fd的最大
        int setsize; /* max number of file descriptors tracked */
        long long timeEventNextId;
        time_t lastTime;     /* Used to detect system clock skew */
        // IO事件数组
        aeFileEvent *events; /* Registered events */
        // 已触发事件数组
        aeFiredEvent *fired; /* Fired events */
        // 记录时间事件的链表头
        aeTimeEvent *timeEventHead;
        int stop;
        // 和API调⽤接⼝相关的数据
        void *apidata; /* This is used for polling API specific data */
        // 进入事件循环流程前执行的函数
        aeBeforeSleepProc *beforesleep;
        // 退出事件循环流程后执行的函数
        aeBeforeSleepProc *aftersleep;
    } aeEventLoop;
    复制代码

### aeCreateEventLoop 函数的初始化操作 ###

我们看一下 `aeCreateEventLoop` 函数的定义:

aeEventLoop *aeCreateEventLoop(int setsize);
    复制代码

可以看到入参只有 `setSize`,在 `sever` 初始化时调用 `aeCreateEventLoop(server.maxclients+CONFIG_FDSET_INCR)`,可以看到 `setSize` 的值是由 `server.maxclients` 和 `CONFIG_FDSET_INCR` 这两个参数决定的，`server.maxclients` 的值可以通过 `redis.conf` 进行定义，默认值为 1000；而宏定义 `CONFIG_FDSET_INCR` 的值是 `CONFIG_MIN_RESERVED_FDS` 加上 96。

#define CONFIG_MIN_RESERVED_FDS 32
    #define CONFIG_FDSET_INCR (CONFIG_MIN_RESERVED_FDS+96)
    复制代码

那么这个参数有什么用呢，我们接下来看一下 `aeCreateEventLoop` 函数的执行流程。

aeEventLoop *aeCreateEventLoop(int setsize) {
        aeEventLoop *eventLoop;
        int i;
        // 分配内存空间
        if ((eventLoop = zmalloc(sizeof(*eventLoop))) == NULL) goto err;
        eventLoop->events = zmalloc(sizeof(aeFileEvent)*setsize);
        eventLoop->fired = zmalloc(sizeof(aeFiredEvent)*setsize);
        if (eventLoop->events == NULL || eventLoop->fired == NULL) goto err;
        eventLoop->setsize = setsize;
        eventLoop->lastTime = time(NULL);
        // 设置时间事件的链表头为NULL
        eventLoop->timeEventHead = NULL;
        eventLoop->timeEventNextId = 0;
        eventLoop->stop = 0;
        eventLoop->maxfd = -1;
        eventLoop->beforesleep = NULL;
        eventLoop->aftersleep = NULL;
        // 调⽤aeApiCreate函数，去实际调⽤操作系统提供的IO多路复⽤函数
        if (aeApiCreate(eventLoop) == -1) goto err;
        /* Events with mask == AE_NONE are not set. So let's initialize the
         * vector with it. */
        // 将所有⽹络IO事件对应⽂件描述符的掩码设置为AE_NONE
        for (i = 0; i < setsize; i++)
            eventLoop->events[i].mask = AE_NONE;
        return eventLoop;
    
    // 如果初始化失败释放内存
    err:
        if (eventLoop) {
            zfree(eventLoop->events);
            zfree(eventLoop->fired);
            zfree(eventLoop);
        }
        return NULL;
    }
    
    复制代码

通过上面的代码分析可知，`aeCreateEventLoop` 首先会创建一个 `aeEventLoop` 结构体类型的变量 `eventLoop`。然后为 `eventLoop` 的成员变量分配内存空间，并且为 `eventLoop` 的成员变量赋初值。第二步调用 `aeApiCreate` 函数去实际调⽤操作系统提供的 `IO` 多路复⽤函数，假设 `Redis` 运⾏在 `Linux` 操作系统上，并且 `IO` 多路复⽤机制是 `epoll`，那么此时，`aeApiCreate` 函数就会调⽤ `epoll_create` 创建 `epoll` 实例，同时会创建 `epoll_event` 结构的数组，数组⼤⼩等于参数 `setsize`。如下面代码所示：

typedef struct aeApiState {
        // epoll实例的描述符
        int epfd; 
        // 记录监听事件
        struct epoll_event *events; 
    } aeApiState;
    
    static int aeApiCreate(aeEventLoop *eventLoop) {
        aeApiState *state = zmalloc(sizeof(aeApiState));
    
        if (!state) return -1;
        // 将epoll_event数组保存在aeApiState结构体变量state中
        state->events = zmalloc(sizeof(struct epoll_event)*eventLoop->setsize);
        if (!state->events) {
            zfree(state);
            return -1;
        }
        // 将epoll实例描述符保存在aeApiState结构体变量state中
        state->epfd = epoll_create(1024); /* 1024 is just a hint for the kernel */
        if (state->epfd == -1) {
            zfree(state->events);
            zfree(state);
            return -1;
        }
        // 把state变量赋值给eventLoop中的apidata
        eventLoop->apidata = state;
        return 0;
    }
    复制代码

最后 `aeCreateEventLoop` 函数会把所有⽹络 `IO` 事件对应⽂件描述符的掩码初始化为 `AE_NONE`，表⽰暂时不对任何事件进行监听。

## IO 事件 ##

### IO 时间创建 ###

首先看一下 `aeCreateFileEvent` 函数的定义：

int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,
            aeFileProc *proc, void *clientData);
    复制代码

`aeCreateFileEvent` 函数有五个参数，分别是循环事件结构体 `*eventLoop`,`IO` 事件对应的文件描述符 `fd`，事件类型掩码 `mask`，事件处理回调函数 `*proc`，以及事件私有数据 `*clientData`。

int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,
            aeFileProc *proc, void *clientData)
    {
        if (fd >= eventLoop->setsize) {
            errno = ERANGE;
            return AE_ERR;
        }
        // 根据事件的 fd 获取事件指针
        aeFileEvent *fe = &eventLoop->events[fd];
    
        // 添加监听的事件事件
        if (aeApiAddEvent(eventLoop, fd, mask) == -1)
            return AE_ERR;
        fe->mask |= mask;
        // 设置回调函数
        if (mask & AE_READABLE) fe->rfileProc = proc;
        if (mask & AE_WRITABLE) fe->wfileProc = proc;
        fe->clientData = clientData;
        if (fd > eventLoop->maxfd)
            eventLoop->maxfd = fd;
        return AE_OK;
    }
    复制代码

## 时间事件 ##

### 时间事件定义 ###

/* Time event structure */
    typedef struct aeTimeEvent {
        // 事件时间 id
        long long id; /* time event identifier. */
        // 事件到达的秒级时间戳
        long when_sec; /* seconds */
        // 事件到达的毫秒级时间戳
        long when_ms; /* milliseconds */
        // 事件触发时的函数
        aeTimeProc *timeProc;
        // 事件结束后的处理函数
        aeEventFinalizerProc *finalizerProc;
        // 事件相关的私有数据
        void *clientData;
        // 前向指针
        struct aeTimeEvent *prev;
        // 后向指针
        struct aeTimeEvent *next;
    } aeTimeEvent;
    复制代码

时间事件结构体中主要的变量，包括以秒记录和以毫秒记录的时间事件触发时的时间戳when\_sec和 when\_ms，以及时间事件触发后的处理函数*timeProc。另外，在时间事件的结构体中，还包含了前向和 后向指针*prev和\*next，这表明时间事件是以链表的形式组织起来的。

### 时间事件创建 ###

long long aeCreateTimeEvent(aeEventLoop *eventLoop, long long milliseconds,
            aeTimeProc *proc, void *clientData,
            aeEventFinalizerProc *finalizerProc)
    {
        // 生成时间事件 id
        long long id = eventLoop->timeEventNextId++;
        aeTimeEvent *te;
        // 分配内存
        te = zmalloc(sizeof(*te));
        if (te == NULL) return AE_ERR;
        te->id = id;
        // 根据 milliseconds 参数计算所创建时间事件具体的触发时间戳
        aeAddMillisecondsToNow(milliseconds,&te->when_sec,&te->when_ms);
        te->timeProc = proc;
        te->finalizerProc = finalizerProc;
        te->clientData = clientData;
        te->prev = NULL;
        te->next = eventLoop->timeEventHead;
        if (te->next)
            te->next->prev = te;
        eventLoop->timeEventHead = te;
        return id;
    }
    复制代码

具体的时间事件创建是在初始化时：

if (aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL) == AE_ERR) {
            serverPanic("Can't create event loop timers.");
            exit(1);
        }
    
    复制代码

可以看到时间事件处理函数是serverCron。

### 时间事件触发 ###

其实，时间事件的检测触发⽐较简单，事件驱动框架的aeMain函数会循环调⽤aeProcessEvents函数，来处 理各种事件。⽽aeProcessEvents函数在执⾏流程的最后，会调⽤processTimeEvents函数处理相应到时 的任务。

/* Check time events */
        if (flags & AE_TIME_EVENTS)
            processed += processTimeEvents(eventLoop);
    复制代码

/* Process time events */
    static int processTimeEvents(aeEventLoop *eventLoop) {
        int processed = 0;
        aeTimeEvent *te;
        long long maxId;
        time_t now = time(NULL);
    
        /* If the system clock is moved to the future, and then set back to the
         * right value, time events may be delayed in a random way. Often this
         * means that scheduled operations will not be performed soon enough.
         *
         * Here we try to detect system clock skews, and force all the time
         * events to be processed ASAP when this happens: the idea is that
         * processing events earlier is less dangerous than delaying them
         * indefinitely, and practice suggests it is. */
        if (now < eventLoop->lastTime) {
            te = eventLoop->timeEventHead;
            while(te) {
                te->when_sec = 0;
                te = te->next;
            }
        }
        eventLoop->lastTime = now;
        //从时间事件链表中取出事件
        te = eventLoop->timeEventHead;
        maxId = eventLoop->timeEventNextId-1;
        while(te) {
            long now_sec, now_ms;
            long long id;
    
            /* Remove events scheduled for deletion. */
            if (te->id == AE_DELETED_EVENT_ID) {
                aeTimeEvent *next = te->next;
                if (te->prev)
                    te->prev->next = te->next;
                else
                    eventLoop->timeEventHead = te->next;
                if (te->next)
                    te->next->prev = te->prev;
                if (te->finalizerProc)
                    te->finalizerProc(eventLoop, te->clientData);
                zfree(te);
                te = next;
                continue;
            }
    
            /* Make sure we don't process time events created by time events in
             * this iteration. Note that this check is currently useless: we always
             * add new timers on the head, however if we change the implementation
             * detail, this check may be useful again: we keep it here for future
             * defense. */
            if (te->id > maxId) {
                te = te->next;
                continue;
            }
            //获取当前时间
            aeGetTime(&now_sec, &now_ms);
            if (now_sec > te->when_sec ||
                (now_sec == te->when_sec && now_ms >= te->when_ms))
            {
                int retval;
    
                id = te->id;
                //调⽤注册的回调函数处理
                retval = te->timeProc(eventLoop, id, te->clientData);
                processed++;
                if (retval != AE_NOMORE) {
                    aeAddMillisecondsToNow(retval,&te->when_sec,&te->when_ms);
                } else {
                    te->id = AE_DELETED_EVENT_ID;
                }
            }
            te = te->next;
        }
        return processed;
    }
    复制代码

## 小结 ##

本文从介绍一条 Redis 命令开始，逐步介绍了事件驱动框架，包括 IO 事件和时间事件。