【源码】Redis 启动与事件循环

本文梳理了 Redis 启动服务器的流程，并且简要介绍了 Redis 事件处理机制的原理以及对应的源码实现。

Redis 作为一个高性能的缓存，必须要有一个高性能的事件处理机制支撑其性能。同时，Redis 的事件处理机制不考虑扩展性（不用于 Redis 以外的应用），保证了实现的简洁性，阅读和分析的难度不大。

本文的 Redis 版本是 redis-6.0.9，源码主要在 server.c 和 ae.c 中。

启动服务器

redis-server 的 main 函数在文件 server.c 中，主要有以下几个步骤：

1. 使用默认参数初始化。
2. 解析命令行参数。
3. 初始化服务器。具体执行了以下的操作：
   • 设置信号处理函数。
   • 初始化事件循环结构体 EventLoop。
   • 把监听套接字作为事件注册到事件循环结构体中。
   • 初始化数据库。
4. 启动服务器的事件循环，开始处理外部的指令。

服务器的事件循环

EventLoop

Redis 把所有的事件注册到一个 aeEventLoop 结构体中，该结构体负责监听是否有事件到来。

redisServer 结构的成员 el 用于事件循环，实现如下：

typedef struct aeEventLoop {
    int maxfd;   /* highest file descriptor currently registered */
    int setsize; /* max number of file descriptors tracked */
    long long timeEventNextId;
    time_t lastTime;     /* Used to detect system clock skew */
    aeFileEvent *events; /* Registered events */
    aeFiredEvent *fired; /* Fired events */
    aeTimeEvent *timeEventHead;
    int stop;
    void *apidata; /* This is used for polling API specific data */
    aeBeforeSleepProc *beforesleep;
    aeBeforeSleepProc *aftersleep;
    int flags;
} aeEventLoop;

server.el 通过函数 aeEventLoop *aeCreateEventLoop(int setsize) 创建，参数 setsize 指定了文件事件列表 aeFileEvent *events 的大小。如果需要销毁 server.el，需要调用函数 void aeDeleteEventLoop(aeEventLoop *eventLoop)。

为了让 Redis 在不同的平台下使用不同的事件处理 api，Redis 把不同平台下的 I/O 多路复用接口封装起来，提供统一的 apiaeApiCreate(eventLoop) 调用。Redis 通过下面的代码确定使用的 I/O 多路复用接口。

#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

例如，Linux 平台下使用 epoll 作为 I/O 多路复用的接口，MacOS 平台使用 kqueue 作为 I/O 多路复用的接口。

server.el 支持两种类型的事件：

• 文件事件 aeFileEvent
• 定时器事件 aeTimeEvent

文件事件

文件事件用于处理网络上的操作，包括连接的建立以及连接的读写。在 aeEventLoop 中通过数组保存。访问的时候通过下标 fd 找到对应的文件事件结构体，实现如下：

typedef struct aeFileEvent {
    int mask; /* one of AE_(READABLE|WRITABLE|BARRIER) */
    aeFileProc *rfileProc;
    aeFileProc *wfileProc;
    void *clientData;
} aeFileEvent;

Redis 通过以下两个函数注册和删除文件事件。

int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask, aeFileProc *proc, void *clientData);
void aeDeleteFileEvent(aeEventLoop *eventLoop, int fd, int mask);

添加事件的时候需要提供一个回调函数，当事件到达时，需要调用对应的回调函数。

和创建 aeEventLoop 相同，文件事件通过 aeApiAddEvent() 和 aeApiDelEvent() 屏蔽底层实现细节，具体使用的多路复用 API 通过操作系统决定。

定时器事件

定时器事件在 aeEventLoop 中以链表的形式保存，结构实现如下：

typedef struct aeTimeEvent {
    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;
    int refcount; /* refcount to prevent timer events from being
  		   * freed in recursive time event calls. */
} aeTimeEvent;

Redis 通过以下的函数实现定时器事件的添加和删除。

long long aeCreateTimeEvent(aeEventLoop *eventLoop, long long milliseconds,
        aeTimeProc *proc, void *clientData,
        aeEventFinalizerProc *finalizerProc);
int aeDeleteTimeEvent(aeEventLoop *eventLoop, long long id);

如果一个定时器事件被触发了，需要调用相应的回调函数 proc()；如果一个定时器事件被取消了，调用相应的回调函数 finalizerProc()。

在服务器初始化的过程中，会注册一个定时器事件 serverCron。根据《Redis 设计与实现》的说法，在 Redis 3.0 仅有这一个定时器事件需要处理，因此采用双向链表实现定时器事件的管理，这样做效率不会下降过多并且保证了实现的简单。

不清楚目前的版本（6.0.9）是否需要处理多个定时器事件，但如果需要呢？Redis 也给出了一种优化思路：采用 skiplist 把插入的事件复杂度降低到 O(log(N))

事件处理

这部分由 aeMain 负责，就是一个循环，整体的流程如下：

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

处理事件由函数 aeProcessEvents() 负责，主要执行了以下步骤：

1. 获得最近的定时器事件，事件复杂度为 O(N)。
2. 计算该事件还需要多久才会被触发，计算多路复用的超时时间。
3. 调用多路复用 API，这个操作会阻塞，直到调用超时或者有新的事件触发。
4. 处理文件事件。
5. 处理定时器事件，这一步通过函数 processTimeEvents() 实现。该函数遍历链表，处理每个到期的定时器事件，并重新添加到定时器事件列表中。

参考

• 《Redis 设计与实现》
• Redis 源码阅读 (一) – eventLoop