[redis 源码走读] - 事件 - 文件事件

redis 服务底层采用了异步事件管理（aeEventLoop）：管理时间事件和文件事件。对大量网络文件描述符（fd）事件管理，redis 建立在安装系统对应的事件驱动基础上（例如 Linux 的 epoll）。

关于事件驱动，本章主要讲述 Linux 系统的 epoll 事件驱动。

关于事件处理，本章主要讲述文件事件，时间事件可以参考帖子《[redis 源码走读] 事件 - 定时器》。

1. 事件驱动

redis 根据安装系统选择对应的事件驱动。

 
    // ae.c
    /* Include the best multiplexing layer supported by this system.
     * The following should be ordered by performances, descending. */
    #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

2. 异步事件管理

epoll 是异步事件驱动，上层逻辑操作和下层事件驱动要通过 fd 文件描述符串联起来。异步事件管理（aeEventLoop），对 epoll 做了一些封装，方便异步事件回调处理。

有关 epoll 工作流程，可以参考我的帖子：epoll 多路复用 I/O工作流程

层次	描述
ae.c	关联异步业务事件和epoll接口，处理fd对应事件逻辑。
ae_epoll.c	对epoll接口进行封装，方便上层操作。
epoll	Linux内核多路复用I/O模型，主要为了高效处理大批量文件描述符事件。

2.1. 数据结构

 
    // ae.c
    
    // 文件事件结构
    typedef struct aeFileEvent {
        int mask; // 事件类型组合（one of AE_(READABLE|WRITABLE|BARRIER)）
        aeFileProc *rfileProc; // 读事件回调操作。
        aeFileProc *wfileProc; // 写事件回调操作。
        void *clientData;      // 业务传入的私有数据。方便回调使用。
    } aeFileEvent;
    
    // 就绪事件
    typedef struct aeFiredEvent {
        int fd;   // 文件描述符。
        int mask; // 事件类型组合。
    } aeFiredEvent;
    
    // 事件管理结构
    typedef struct aeEventLoop {
        int maxfd;   // 监控的最大文件描述符。
        int setsize; // 处理文件描述符个数。
        ...
        aeFileEvent *events; // 根据 fd 监听事件。
        aeFiredEvent *fired; // 从内核取出的就绪事件。
        ...
    } aeEventLoop;

结构	描述
aeEventLoop	文件事件和时间事件管理。
aeFileEvent	文件事件结构，方便异步回调逻辑调用。aeEventLoop会创建一个aeFileEvent数组，数组下标是fd，fd对应aeFileEvent数据结构。
aeFiredEvent	从内核获取的就绪事件。（例如Linux系统通过epoll_wait接口获取就绪事件，每个事件分别存储在aeFiredEvent数组中）

2.2. 创建事件管理对象

创建事件管理对象，对监控的文件数量设置了上限。

文件监控上限配置。

 
    # redis.conf
    #
    # Set the max number of connected clients at the same time. By default
    # this limit is set to 10000 clients, however if the Redis server is not
    # able to configure the process file limit to allow for the specified limit
    # the max number of allowed clients is set to the current file limit
    # minus 32 (as Redis reserves a few file descriptors for internal uses).
    #
    # Once the limit is reached Redis will close all the new connections sending
    # an error 'max number of clients reached'.
    #
    # maxclients 10000

创建事件管理对象。

 
    #define CONFIG_MIN_RESERVED_FDS 32
    #define CONFIG_FDSET_INCR (CONFIG_MIN_RESERVED_FDS+96)
    
    // server.c
    void initServer(void) {
        ...
        server.el = aeCreateEventLoop(server.maxclients+CONFIG_FDSET_INCR);
        ...
    }
    
    int main(int argc, char **argv) {
        ...
        initServer();
        ...
    }

2.3. 事件处理流程

循环处理事件

 
    // server.c
    int main(int argc, char **argv) {
        ...
        aeMain(server.el);
        ...
    }
    
    // ae.c
    // 循环处理事件
    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);
        }
    }

添加事件，关联 fd 事件与异步回调相关信息。

 
    int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,
            aeFileProc *proc, void *clientData) {
        if (fd >= eventLoop->setsize) {
            errno = ERANGE;
            return AE_ERR;
        }
        aeFileEvent *fe = &eventLoop->events[fd];
    
        // 调用底层 epoll_ctl 注册事件。
        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;
    }

删除事件，删除对应 fd 的事件。

 
    void aeDeleteFileEvent(aeEventLoop *eventLoop, int fd, int mask) {
        if (fd >= eventLoop->setsize) return;
        aeFileEvent *fe = &eventLoop->events[fd];
        if (fe->mask == AE_NONE) return;
    
        // 如果删除的是写事件，要把写事件优先处理的事件也去掉，恢复优先处理读事件，再处理写事件逻辑。
        if (mask & AE_WRITABLE) mask |= AE_BARRIER;
    
        // 调用底层 epoll_ctl 修改删除事件。
        aeApiDelEvent(eventLoop, fd, mask);
        fe->mask = fe->mask & (~mask);
        if (fd == eventLoop->maxfd && fe->mask == AE_NONE) {
            /* Update the max fd */
            int j;
    
            for (j = eventLoop->maxfd-1; j >= 0; j--)
                if (eventLoop->events[j].mask != AE_NONE) break;
            eventLoop->maxfd = j;
        }
    }

2.4. 事件处理逻辑

文件事件处理逻辑，从内核取出就绪事件，根据事件的读写类型，分别进行回调处理相关业务逻辑。

 
    // ae.c
    int aeProcessEvents(aeEventLoop *eventLoop, int flags) {
        ...
        // 多路复用接口，从内核取出就绪事件。
        numevents = aeApiPoll(eventLoop, tvp);
        ...
        for (j = 0; j < numevents; j++) {
            // 根据就绪事件 fd，取出对应的异步文件事件进行逻辑处理。
            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. */
    
            /* AE_BARRIER 表示优先可写事件。正常情况，一般先读后写。
             * AE_BARRIER 使用场景，有兴趣的朋友，可以查找源码关键字：CONN_FLAG_WRITE_BARRIER
             * 理解这部分的逻辑。 */
            int invert = fe->mask & AE_BARRIER;
    
            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++;
                }
            }
            ...
        }
        ...
    }

2.5. 获取待处理事件

通过 epoll_wait 从系统内核取出就绪文件事件进行处理。

 
    // ae_epoll.c
    static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
        aeApiState *state = eventLoop->apidata;
        int retval, numevents = 0;
    
        // 从内核取出就绪文件事件进行处理。
        retval = epoll_wait(state->epfd,state->events,eventLoop->setsize,
                tvp ? (tvp->tv_sec*1000 + tvp->tv_usec/1000) : -1);
        if (retval > 0) {
            int j;
    
            numevents = retval;
            for (j = 0; j < numevents; j++) {
                int mask = 0;
                struct epoll_event *e = state->events+j;
    
                if (e->events & EPOLLIN) mask |= AE_READABLE;
                if (e->events & EPOLLOUT) mask |= AE_WRITABLE;
                if (e->events & EPOLLERR) mask |= AE_WRITABLE|AE_READABLE;
                if (e->events & EPOLLHUP) mask |= AE_WRITABLE|AE_READABLE;
    
                // 就绪事件和fd保存到 fired。
                eventLoop->fired[j].fd = e->data.fd;
                eventLoop->fired[j].mask = mask;
            }
        }
        return numevents;
    }

3. 总结

redis 没有使用第三方库，实现跨平台的异步事件驱动。对文件事件驱动封装也比较简洁高效。

4. 参考

Java 面试宝典是大明哥全力打造的 Java 精品面试题，它是一份靠谱、强大、详细、经典的 Java 后端面试宝典。它不仅仅只是一道道面试题，而是一套完整的 Java 知识体系，一套你 Java 知识点的扫盲贴。

它的内容包括：

大厂真题：Java 面试宝典里面的题目都是最近几年的高频的大厂面试真题。
原创内容：Java 面试宝典内容全部都是大明哥原创，内容全面且通俗易懂，回答部分可以直接作为面试回答内容。
持续更新：一次购买，永久有效。大明哥会持续更新 3+ 年，累计更新 1000+，宝典会不断迭代更新，保证最新、最全面。
覆盖全面：本宝典累计更新 1000+，从 Java 入门到 Java 架构的高频面试题，实现 360° 全覆盖。
不止面试：内容包含面试题解析、内容详解、知识扩展，它不仅仅只是一份面试题，更是一套完整的 Java 知识体系。
宝典详情：https://www.yuque.com/chenssy/sike-java/xvlo920axlp7sf4k
宝典总览：https://www.yuque.com/chenssy/sike-java/yogsehzntzgp4ly1
宝典进展：https://www.yuque.com/chenssy/sike-java/en9ned7loo47z5aw

目前 Java 面试宝典累计更新 400+ 道，总字数 42w+。大明哥还在持续更新中，下图是大明哥在 2024-12 月份的更新情况：

想了解详情的小伙伴，扫描下面二维码加大明哥微信【daming091】咨询

同时，大明哥也整理一套目前市面最常见的热点面试题。微信搜[大明哥聊 Java]或扫描下方二维码关注大明哥的原创公众号[大明哥聊 Java] ，回复【面试题】即可免费领取。

阅读全文