夜雨聆风在之前的项目中简单实现了一个事件循环，趁此机会学习了一下优秀开源实现：Redis的事件库ae。在这里做一些笔记
开源代码路径：
https://github.com/redis/redis/blob/unstable/src/ae.c
https://github.com/redis/redis/blob/unstable/src/ae.h
https://github.com/redis/redis/blob/unstable/src/ae_epoll.c
一、引言
Redis（Remote Dictionary Server）是一个开源的、基于内存的高性能键值对数据库。与传统将数据存储在磁盘上的关系型数据库不同，Redis将所有数据驻留在内存中，从而实现了微秒级的读写延迟和单机十万级QPS（Queries Per Second）的吞吐能力。
Redis原生支持字符串、哈希、列表、集合、有序集合、流等多种数据结构，并内置了持久化、主从复制、哨兵自动故障转移和集群水平扩展等企业级特性，能够胜任缓存加速、分布式锁、消息队列、实时排行榜、会话管理等丰富场景。在现代后端架构中，Redis几乎总是与MySQL等关系型数据库配合使用：后者保障数据的持久安全与复杂查询，前者则扛住高并发访问与实时性要求，两者共同构成了当今互联网应用的数据层基石。
Redis的高性能来源于“纯内存操作”和“单线程模型”：内存消除了磁盘 IO 的物理瓶颈，单线程避免了多线程锁竞争与上下文切换的开销但这。但是一个线程究竟如何在不阻塞的前提下同时感知并处理成百上千个并发连接？Redis没有使用libevent库，而是针对自身使用场景实现了一个事件循环库：ae（Async Event Loop），这是Redis单线程Reactor架构的核心。
本文通过几个ae.h，ae.c和ae_epoll.c这几个ae库文件，浅浅分析一下ae事件库如何高效处理并发连接
二、ae提供的API
ae.h中声明了如下接口。其中支持注册文件事件和定时器事件。
1/* Prototypes */
2aeEventLoop *aeCreateEventLoop(int setsize); // 创建事件循环，指定初始最大连接数量
3voidaeDeleteEventLoop(aeEventLoop *eventLoop); // 销毁事件循环
4voidaeStop(aeEventLoop *eventLoop); // 结束事件循环
5
6intaeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask, aeFileProc *proc, void *clientData); // 注册一个文件事件
7voidaeDeleteFileEvent(aeEventLoop *eventLoop, int fd, int mask); // 注销一个文件事件
8intaeGetFileEvents(aeEventLoop *eventLoop, int fd); // 获取已注册文件事件的事件掩码
9void *aeGetFileClientData(aeEventLoop *eventLoop, int fd); // 获取已注册文件事件绑定的用户参数
10
11longlongaeCreateTimeEvent(aeEventLoop *eventLoop, longlong milliseconds,aeTimeProc *proc, void *clientData, aeEventFinalizerProc *finalizerProc); // 创建一个定时事件
12intaeDeleteTimeEvent(aeEventLoop *eventLoop, longlong id); // 删除指定定时事件
13
14intaeProcessEvents(aeEventLoop *eventLoop, int flags); // 执行事件处理的核心函数
15intaeWait(int fd, int mask, longlong milliseconds); // 对单个fd进行同步阻塞等待
16voidaeMain(aeEventLoop *eventLoop); // 启动事件循环的入口
17
18char *aeGetApiName(void); // 获取底层IO复用的API名称，例如epoll kqueue evport select
19
20voidaeSetBeforeSleepProc(aeEventLoop *eventLoop, aeBeforeSleepProc *beforesleep); // 注册休眠前钩子函数
21voidaeSetAfterSleepProc(aeEventLoop *eventLoop, aeBeforeSleepProc *aftersleep); // 注册唤醒后的钩子函数
22
23intaeGetSetSize(aeEventLoop *eventLoop); // 获取当前事件循环的最大fd容量
24intaeResizeSetSize(aeEventLoop *eventLoop, int setsize); // 动态调整事件循环最大fd容量
25
26voidaeSetDontWait(aeEventLoop *eventLoop, int noWait); // 设置事件循环为非阻塞轮询
从使用者的角度看，完整使用流程：
通过 aeCreateEventLoop创建事件循环通过 aeCreateFileEvent/aeCreateTimeEvent注册事件调用 aeMain启动主循环使用 aeStop结束事件循环
三、核心数据结构
3.1 aeEventLoop
aeEventLoop结构体包含了事件循环的全部状态。
其中events数组按照fd索引，支持随机查找事件事件，但限制fd必须小于setsize，考虑到Redis日常处理的连接数不多并且fd较为紧凑，使用数组优于使用哈希表存储
1typedefstructaeEventLoop {
2int maxfd; // 当前注册的最大fd
3int setsize; // 最大支持fd数量
4longlong timeEventNextId; // 时间事件自增ID
5int nevents; // 当前events数组容量
6 aeFileEvent *events; // 文件事件数组，以fd作为索引
7 aeFiredEvent *fired; // 就绪事件数组，快照
8 aeTimeEvent *timeEventHead; // 时间事件链表
9int stop; // 循环停止标志，用于优雅关闭
10void *apidata; // 多路复用层私有数据（如 epoll fd）
11 aeBeforeSleepProc *beforesleep; // 休眠前钩子函数
12 aeBeforeSleepProc *aftersleep; // 唤醒后钩子函数
13int flags; // 控制标志
14void *privdata[2]; // 用户私有数据
15} aeEventLoop;
3.2 aeFileEvent
aeFileEvent是对文件事件的描述，其中包含一个mask标记当前关注的事件类型，以及读写回调和传给回调的用户数据
1#define AE_NONE 0 /* No events registered. */
2#define AE_READABLE 1 /* Fire when descriptor is readable. */
3#define AE_WRITABLE 2 /* Fire when descriptor is writable. */
4#define AE_BARRIER 4 /* With WRITABLE, never fire the event if the
5 READABLE event already fired in the same event
6 loop iteration. Useful when you want to persist
7 things to disk before sending replies, and want
8 to do that in a group fashion. */
9
10typedefstructaeFileEvent {
11int mask; // 关注事件类型掩码
12 aeFileProc *rfileProc; // 读回调
13 aeFileProc *wfileProc; // 写回调
14void *clientData; // 用户数据
15} aeFileEvent;
3.3 aeTimeEvent
aeTimeEvent描述时间事件
1typedefstructaeTimeEvent {
2longlong id; // 时间事件id
3 monotime when; // 绝对触发时间
4 aeTimeProc *timeProc; // 到期回调函数
5 aeEventFinalizerProc *finalizerProc;
6void *clientData; // 用户数据
7structaeTimeEvent *prev, *next;// 双向链表前驱后继
8int refcount; // 引用计数
9} aeTimeEvent;
四、多路复用
ae的一大设计亮点是对底层多路复用API的封装，在ae.c文件开头，通过条件编译选择最优实现：
1#ifdef HAVE_EVPORT
2#include"ae_evport.c"
3#else
4#ifdef HAVE_EPOLL
5#include"ae_epoll.c"
6#else
7#ifdef HAVE_KQUEUE
8#include"ae_kqueue.c"
9#else
10#include"ae_select.c"
11#endif
12#endif
13#endif
每类实现需要提供统一的函数接口：
aeApiCreate：初始化 aeApiResize：调整事件数组大小 aeApiFree：释放资源 aeApiAddEvent/ aeApiDelEvent：增删事件aeApiPoll：等待事件就绪
以epoll为例，实现位于ae_epoll.c，其中aeApiCreate创建epoll fd，使用的是默认LT模式，分配相关内存；aeApiResize调用zrealloc（Redis的内存分配器）扩容数组；aeApiFree关闭epoll fd，释放内存；aeApiAddEvent根据事件mask选择EPOLL_CTL_ADD/EPOLL_CTL_MOD，添加/修改epoll监听EPOLLIN/EPOLLOUT事件；aeApiDelEvent同理，取消监听一类事件或者从epoll中注销；aeApiPoll进行epoll_wait，等待事件就绪
五、文件事件的注册/删除
5.1 注册：`aeCreateFileEvent`
函数首先检查fd是否超出 setsize，若超出则返回错误
接着有一个动态扩容的机制：
1/**
2* Resize the events and fired arrays if the file
3* descriptor exceeds the current number of events.
4*/
5if (unlikely(fd >= eventLoop->nevents)) {
6 int newnevents = eventLoop->nevents;
7 newnevents = (newnevents * 2 > fd + 1) ? newnevents * 2 : fd + 1;
8 newnevents = (newnevents > eventLoop->setsize) ? eventLoop->setsize : newnevents;
9 eventLoop->events = zrealloc(...);
10 eventLoop->fired = zrealloc(...);
11for (int i = eventLoop->nevents; i < newnevents; i++)
12 eventLoop->events[i].mask = AE_NONE;
13 eventLoop->nevents = newnevents;
14}
其中nevents初始为min(setsize, 1024)，当你注册一个更大的fd时，它会自动扩容到刚好容纳该fd，且不超过 setsize。节省了内存，避免了频繁 realloc。
然后调用aeApiAddEvent通知底层多路复用相应操作，最后修改events数组对应fd槽位的回调函数、参数等，更新maxFd
5.2 删除：`aeDeleteFileEvent`
1aeFileEvent *fe = &eventLoop->events[fd];
2if (fe->mask == AE_NONE) return;
3
4/* We want to always remove AE_BARRIER if set when AE_WRITABLE
5 * is removed. */
6if (mask & AE_WRITABLE) mask |= AE_BARRIER;
7
8aeApiDelEvent(eventLoop, fd, mask);
9fe->mask = fe->mask & (~mask);
10if (fd == eventLoop->maxfd && fe->mask == AE_NONE) {
11/* Update the max fd */
12 int j;
13
14for (j = eventLoop->maxfd-1; j >= 0; j--)
15if (eventLoop->events[j].mask != AE_NONE) break;
16 eventLoop->maxfd = j;
17}
删除时如果移除的是 AE_WRITABLE，同时移除 AE_BARRIER（这个屏障用于设置先写后读）。接着调用 aeApiDelEvent，更新 events数组。
如果该fd是当前最大且事件全空，那么修改maxFd，对于maxfd的维护对于select实现是有用的（select需要遍历到），但对于epoll其实可有可无
六、时间事件：链表 + 惰性删除
时间事件使用双向链表，新事件总是插在头部。查询最近到期时间需要遍历整个链表，复杂度O(N)。ae.c中获取最近到期定时器的接口有注释说明存在优化空间：
1/* How many microseconds until the first timer should fire.
2 * If there are no timers, -1 is returned.
3 *
4 * Note that's O(N) since time events are unsorted.
5 * Possible optimizations (not needed by Redis so far, but...):
6 * 1) Insert the event in order, so that the nearest is just the head.
7 * Much better but still insertion or deletion of timers is O(N).
8 * 2) Use a skiplist to have this operation as O(1) and insertion as O(log(N)).
9 */
10static int64_t usUntilEarliestTimer(aeEventLoop *eventLoop){...}
主流的用户态定时器一般使用最小堆管理，支持大量定时器场景下O(logN)查找最近到期定时器。ae考虑到Redis使用的定时器不多因而没有实现
6.1 创建定时事件：`aeCreateTimeEvent`
创建定时接口中，分配内存，然后计算绝对时间（getMonotonicUs() + milliseconds*1000），头插到双向链表中。
6.2 删除定时事件：`aeDeleteTimeEvent`
删除定时事件，时只是将对应id标记为AE_DELETED_EVENT_ID，真正的释放发生在processTimeEvents中
这种惰性删除的设计，避免了在删除时持有锁或影响遍历。
6.3 处理定时事件：`processTimeEvents`
处理定时事件时，遍历链表，对于已标记删除的节点，如果引用计数refcount == 0，则释放并调用finalizerProc回调。
对于未删除且when <= now的到期事件，调用timeProc，根据返回值重新调度：
返回 AE_NOMORE：标记为删除（延迟释放）返回正值（毫秒）：更新 when = now + retval * 1000，实现周期性定时
注意refcount的用法：在调用 timeProc 前 refcount++，调用后 refcount--。这是为了防止在timeProc内部又调用了aeDeleteTimeEvent 导致节点被提前释放（递归调用场景）。
1if (te->when <= now) {
2 int retval;
3
4 id = te->id;
5 te->refcount++;
6 retval = te->timeProc(eventLoop, id, te->clientData);
7 te->refcount--;
8 processed++;
9 now = getMonotonicUs();
10if (retval != AE_NOMORE) {
11 te->when = now + (monotime)retval * 1000;
12 } else {
13 te->id = AE_DELETED_EVENT_ID;
14 }
15}
七、主事件循环：`aeProcessEvents`
aeProcessEvents是核心调度器的实现，在aeMain中循环调用，直到设置stop标志退出。
1voidaeMain(aeEventLoop *eventLoop){
2 eventLoop->stop = 0;
3while (!eventLoop->stop) {
4 aeProcessEvents(eventLoop, AE_ALL_EVENTS|
5 AE_CALL_BEFORE_SLEEP|
6 AE_CALL_AFTER_SLEEP);
7 }
8}
其中负责：
- 根据标志决定是否等待文件事件。
- 计算等待超时（到最近的时间事件）。
调用 aeApiPoll阻塞。- 处理就绪的文件事件（按顺序）。
- 处理时间事件。
- 调用 before/after _sleep钩子。
7.1 等待策略
1if (eventLoop->maxfd != -1 || ((flags & AE_TIME_EVENTS) && !(flags & AE_DONT_WAIT))) {
2// 计算 tvp
3if ((flags & AE_DONT_WAIT) || (eventLoop->flags & AE_DONT_WAIT)) {
4 tv.tv_sec = tv.tv_usec = 0; // 非阻塞轮询
5 tvp = &tv;
6 } elseif (flags & AE_TIME_EVENTS) {
7 usUntilTimer = usUntilEarliestTimer(eventLoop);
8if (usUntilTimer >= 0) { ... } // 设置超时
9 }
10 numevents = aeApiPoll(eventLoop, tvp);
11}
等待时，若存在定时事件，则获取最近到期的定时器，设置相应的等待事件给aeApiPoll，否则底层阻塞等待
如果没有任何事件，aeProcessEvents会直接返回0。aeMain会重复调用
7.2 文件事件触发顺序
对于每一个就绪的 fd，可能会有读和写同时就绪。默认顺序是先读后写（因为读请求通常需要处理，写回复可以稍后）
但Redis有时需要在回复客户端之前做一些持久化操作，比如刷新数据到盘内，这时希望先写后读，于是引入了AE_BARRIER标志
1/* Normally we execute the readable event first, and the writable
2 * event later. This is useful as sometimes we may be able
3 * to serve the reply of a query immediately after processing the
4 * query.
5 *
6 * However if AE_BARRIER is set in the mask, our application is
7 * asking us to do the reverse: never fire the writable event
8 * after the readable. In such a case, we invert the calls.
9 * This is useful when, for instance, we want to do things
10 * in the beforeSleep() hook, like fsyncing a file to disk,
11 * before replying to a client. */
12int invert = fe->mask & AE_BARRIER;
13
14/* Note the "fe->mask & mask & ..." code: maybe an already
15 * processed event removed an element that fired and we still
16 * didn't processed, so we check if the event is still valid.
17 *
18 * Fire the readable event if the call sequence is not
19 * inverted. */
20if (!invert && fe->mask & mask & AE_READABLE) {
21 fe->rfileProc(eventLoop,fd,fe->clientData,mask);
22 fired++;
23 fe = &eventLoop->events[fd]; /* Refresh in case of resize. */
24}
25
26/* Fire the writable event. */
27if (fe->mask & mask & AE_WRITABLE) {
28if (!fired || fe->wfileProc != fe->rfileProc) {
29 fe->wfileProc(eventLoop,fd,fe->clientData,mask);
30 fired++;
31 }
32}
33
34/* If we have to invert the call, fire the readable event now
35 * after the writable one. */
36if (invert) {
37 fe = &eventLoop->events[fd]; /* Refresh in case of resize. */
38if ((fe->mask & mask & AE_READABLE) &&
39 (!fired || fe->wfileProc != fe->rfileProc))
40 {
41 fe->rfileProc(eventLoop,fd,fe->clientData,mask);
42 fired++;
43 }
44}
当 fe->mask & AE_BARRIER 为真时，调换顺序：
- 先执行写回调（如果就绪）。
- 再执行读回调（如果就绪且未被写回调覆盖）。
7.3 回调执行的防护
在回调执行过程中，可能会修改events数组（如删除或添加事件），为此ae使用一个fired作为快照，在aeApiPoll中进行赋值，其中包含就绪的fd和mask
1/* A fired event */
2typedefstructaeFiredEvent {
3int fd;
4int mask;
5} aeFiredEvent;
6
7intaeProcessEvents(aeEventLoop *eventLoop, int flags)
8{
9 ...
10
11int fd = eventLoop->fired[j].fd;
12 aeFileEvent *fe = &eventLoop->events[fd];
13int mask = eventLoop->fired[j].mask;
14int fired = 0; /* Number of events fired for current fd. */
15
16if (!invert && fe->mask & mask & AE_READABLE) {
17 fe->rfileProc(eventLoop,fd,fe->clientData,mask);
18 fired++;
19 fe = &eventLoop->events[fd]; /* Refresh in case of resize. */
20 }
21
22 ...
23}
在每次回调后都重新获取 fe = &eventLoop->events[fd]，结合fired数组，确保指针有效，比如读回调需要满足fe->mask & mask & AE_READABLE才会触发。
同时，如果读回调已经触发且写回调与读回调是同一个函数，则避免重复调用（if (!fired || fe->wfileProc != fe->rfileProc)）
八、总结
总结ae的设计，可以看到它没有追求通用事件库的大而全，而是专门以服务Redis单线程Reactor模型而设计
- 放弃哈希表或平衡树，直接以fd索引数组实现O(1)查找，这是建立在Redis连接数量可控且fd分布紧凑的前提下。以少量内存开销换取零哈希冲突、指针跳转，同事支持动态扩容避免预分配浪费内存
- 放弃最小堆，使用O(N)链表，这是建立在Redis活跃的定时器数量极少的情况下，维护链表远比维护堆开销小。可见理论最优不等于工程最优解，数据结构选择不应脱离实际负载特征。
- 多处防御性编程，例如引用计数，fired数组快照隔离。实现单线程模型下，一个回调不会破坏整个事件循环
- 平台层抽象，通过条件编译实现多路复用的底层切换