5.3 源码阅读：Redis ae 事件循环

在许多复杂的网络服务架构中，事件驱动库往往伴随着繁复的线程池封装、平台抽象适配与深层回调队列。然而 Redis 自带的事件循环库 ae（A simple event-driven library）却走向了极致的精简主义：其核心驱动代码仅有约 500 行 C 语言实现，即便加上 Linux epoll、macOS kqueue 等底层多路复用适配层，整体规模也不过 700 行上下。正是这套结构极简、完全摒弃过度设计的单线程事件引擎，支撑起了整个 Redis 服务器每秒数十万次的高吞吐网络 I/O 与毫秒级定时任务调度。

本节研读基于 Redis 稳定版本的 src/ae.h、src/ae.c 与 src/ae_epoll.c 等文件。软件许可证提供 RSALv2 / SSPLv1 / AGPLv3 供选择。

1. 核心数据结构

ae 库将整个事件循环运行时的生命周期状态收敛在一个紧凑的 aeEventLoop 结构体中：

typedef struct aeEventLoop {
    int maxfd;                   /* 当前注册的最大文件描述符 */
    int setsize;                 /* 能跟踪的 fd 上限 */
    long long timeEventNextId;   /* 下一个时间事件的 ID */
    int nevents;                 /* events 与 fired 数组当前的长度 */
    aeFileEvent *events;         /* 按 fd 索引的文件事件数组 */
    aeFiredEvent *fired;         /* 本轮 poll 返回的就绪事件 */
    aeTimeEvent *timeEventHead;  /* 时间事件链表头 */
    int stop;                    /* 非零则退出主循环 */
    void *apidata;               /* 平台 I/O 多路复用的私有数据 */
    aeBeforeSleepProc *beforesleep;
    aeBeforeSleepProc *aftersleep;
    int flags;
    void *privdata[3];
} aeEventLoop;

该结构体在逻辑上统一调度两类不同维度的驱动信号：文件事件与时间事件。

文件事件：fd + mask + 回调

文件事件用 aeFileEvent 表示，每个文件描述符对应数组中的一个槽位：

typedef struct aeFileEvent {
    int mask;              /* AE_READABLE | AE_WRITABLE | AE_BARRIER */
    aeFileProc *rfileProc; /* 可读时的回调 */
    aeFileProc *wfileProc; /* 可写时的回调 */
    void *clientData;      /* 传给回调的用户数据 */
} aeFileEvent;

在设计上，events 数组直接采用文件描述符 fd 作为下标进行直接寻址，events[fd] 即可精确定位该套接字所绑定的事件信息。这种以空间换时间的设计完全省去了哈希散列计算或树形检索的开销，确保了每次事件查询均拥有极为稳定的 O(1) 性能。其折中代价在于预分配数组的大小严格受限于 setsize（在 Redis 中通常预设为 maxclients + 128）。

时间事件：链表 + 单调时钟

时间事件用双向链表串联，每个节点记录触发时刻和回调：

typedef struct aeTimeEvent {
    long long id;
    monotime when;                  /* 触发时刻（微秒级单调时钟） */
    aeTimeProc *timeProc;           /* 回调函数 */
    aeEventFinalizerProc *finalizerProc;
    void *clientData;
    struct aeTimeEvent *prev;
    struct aeTimeEvent *next;
    int refcount;                   /* 防止递归调用时提前释放 */
} aeTimeEvent;

与海量的套接字文件事件不同，Redis 服务端在生命周期中常驻的时间事件极少（在绝大部分生产场景下仅有 serverCron 这一个全局时钟心跳），因此基于双向链表的 O(N) 线性遍历开销几乎可以忽略不计。源码注释中也探讨过引入跳表或最小堆进行有序优化的可能性，但同时明确指出基于 Redis 自身的实际运行负荷，极简的双向链表已是工程性价比最高且易于维护的选择。

2. 主循环：aeMain 与 aeProcessEvents

驱动 Redis 运行的核心引擎在最外层呈现为一个结构极其精悍的顶层循环：

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

整个事件调度的核心控制逻辑均被收敛在 aeProcessEvents 函数内。在循环的单次迭代中，系统依次推进四个阶段的动作：

aeProcessEvents 的四步流程

调用 beforesleep 钩子：Redis 在这里向客户端写出待发送的回复、刷写 AOF 缓冲区等。
计算 poll 超时：遍历时间事件链表，找到最近一个定时器的剩余时间，作为 aeApiPoll 的超时参数。没有定时器则无限等待；设了 AE_DONT_WAIT 则超时为 0（立即返回）。
调用 aeApiPoll 阻塞等待：底层调用 epoll_wait / kevent / select，返回就绪的文件描述符。
分发事件：先遍历就绪的文件事件并调用回调，再处理到期的时间事件。

超时计算的核心是 usUntilEarliestTimer：

static int64_t usUntilEarliestTimer(aeEventLoop *eventLoop) {
    aeTimeEvent *te = eventLoop->timeEventHead;
    if (te == NULL) return -1;

    aeTimeEvent *earliest = NULL;
    while (te) {
        if ((!earliest || te->when < earliest->when) && te->id != AE_DELETED_EVENT_ID)
            earliest = te;
        te = te->next;
    }
    if (earliest == NULL) return -1;

    monotime now = getMonotonicUs();
    return (now >= earliest->when) ? 0 : earliest->when - now;
}

在计算等待时间跨度时，系统底层严格采用单调时钟（getMonotonicUs）而非容易受外部干扰的日历时钟（如 gettimeofday）。日历时钟在发生管理员手动校时或 NTP 时间步进同步时会出现向前跳变或回拨，极易导致定时器陷入长时间睡眠或瞬间被大量错误触发；而单调时钟从操作系统引导起线性递增，为定时事件提供了坚实的时间基准保证。

3. 文件事件的注册与分发

3.1 注册

当网络层需要对特定套接字开启数据监听时，统一调用 aeCreateFileEvent 完成读写事件与回调函数的挂载：

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

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

值得注意的代码细节是：fe->mask |= mask 采用按位或运算来累加监听掩码，这意味着单个文件描述符可以同时独立注册可读与可写事件，并各自挂接不同的回调逻辑。辅助函数 aeApiAddEvent 会自动感知该描述符当前的登记状态：对于初次加入的描述符触发 epoll_ctl(EPOLL_CTL_ADD)，而对于已在监听集合中仅做事件变更的描述符则平滑转为 epoll_ctl(EPOLL_CTL_MOD)。

3.2 分发与 AE_BARRIER

文件事件的分发集中在 aeProcessEvents 的核心循环中。当底层多路复用系统调用（如 epoll_wait）返回就绪的文件描述符后，循环逐一取出已触发的事件并派发给注册的回调：

for (j = 0; j < numevents; j++) {
    int fd = eventLoop->fired[j].fd;
    aeFileEvent *fe = &eventLoop->events[fd];
    int mask = eventLoop->fired[j].mask;
    int fired = 0; /* 当前 fd 已触发的回调数 */

    int invert = fe->mask & AE_BARRIER;

    /* 正常顺序：先读后写 */
    if (!invert && fe->mask & mask & AE_READABLE) {
        fe->rfileProc(eventLoop, fd, fe->clientData, mask);
        fired++;
        fe = &eventLoop->events[fd]; /* 数组可能已扩容，重新取 */
    }

    if (fe->mask & mask & AE_WRITABLE) {
        if (!fired || fe->wfileProc != fe->rfileProc) {
            fe->wfileProc(eventLoop, fd, fe->clientData, mask);
            fired++;
        }
    }

    /* AE_BARRIER：先写后读 */
    if (invert) {
        fe = &eventLoop->events[fd]; /* 数组可能已扩容，重新取 */
        if ((fe->mask & mask & AE_READABLE) &&
            (!fired || fe->wfileProc != fe->rfileProc))
        {
            fe->rfileProc(eventLoop, fd, fe->clientData, mask);
            fired++;
        }
    }

    processed++;
}

在常规的网络服务模型下，事件分发默认遵循“先读后写”的顺序。服务端首先读取套接字缓冲区中客户端送来的请求命令，在内存中执行完毕并将生成的响应载入写缓冲区；紧接着若套接字具备写就绪条件，就可以在同一个循环周期内顺道将响应数据写回网络，从而省去等待下一次轮询的延迟。

然而在需要保证落盘持久化的严苛场景下，这一顺序需要反转。比如某些模块或事务处理要求先通过写回调把应答数据或同步日志 fsync 到磁盘，确认持久化成功后，才允许继续读取该连接后续排队的新命令。标志位 AE_BARRIER 正是为此而设：当掩码中设置了 AE_BARRIER，invert 标志为真，主循环便会打破默认顺序，转为“先写后读”，阻断未落盘前读取后续命令的可能性。

这段分发逻辑里还有两处耐人寻味的实现细节：

一是在调用回调函数之后，代码立即执行了 fe = &eventLoop->events[fd] 重新获取指针。这是因为回调函数内部可能调用了 aeResizeSetSize 等管理接口，导致事件表数组发生了内存重分配（zrealloc）；此时旧的指针可能已经悬空，重新寻址能够确保后续操作基于最新的有效内存地址。

二是 fe->wfileProc != fe->rfileProc 这重判断条件。在某些特定抽象下，读回调与写回调可能被注册为同一个函数入口。增加这一校验并配合 fired 计数器，能够保证在同一个就绪轮次中，同一个函数不会因同一次就绪状态被无意调用两次。

4. 时间事件的管理

4.1 创建

long long aeCreateTimeEvent(aeEventLoop *eventLoop, long long milliseconds,
        aeTimeProc *proc, void *clientData,
        aeEventFinalizerProc *finalizerProc)
{
    long long id = eventLoop->timeEventNextId++;
    aeTimeEvent *te = zmalloc(sizeof(*te));

    te->id = id;
    te->when = getMonotonicUs() + milliseconds * 1000;
    te->timeProc = proc;
    te->finalizerProc = finalizerProc;
    te->clientData = clientData;
    te->prev = NULL;
    te->next = eventLoop->timeEventHead;
    te->refcount = 0;
    if (te->next)
        te->next->prev = te;
    eventLoop->timeEventHead = te;
    return id;
}

新事件插入链表头部，时间复杂度为 $O (1)$ 。这里调用方传入的 milliseconds 是相对时间差（即“从当前起算多少毫秒后触发”），函数内部通过单调时钟读取当前绝对时刻并完成换算，将其存储为微秒精度的绝对时间戳。

4.2 处理与重调度

processTimeEvents 遍历时间事件链表，逐个检查并执行所有到期的事件：

if (te->when <= now) {
    int retval;

    id = te->id;
    te->refcount++;
    retval = te->timeProc(eventLoop, id, te->clientData);
    te->refcount--;
    processed++;
    now = getMonotonicUs();                         /* 回调可能耗时，刷新当前时间 */
    if (retval != AE_NOMORE) {
        te->when = now + (monotime)retval * 1000;   /* 重调度 */
    } else {
        te->id = AE_DELETED_EVENT_ID;               /* 标记删除 */
    }
}

定时器回调函数的返回值直接决定了该事件后续的调度行为：若返回一个正整数，代表这是一个周期性任务，数值即下一次触发与当前时刻的时间间隔（毫秒），主循环据此直接更新 te->when；若返回 AE_NOMORE（即 -1），则表明这是一次性任务，执行完毕后即可结束生命周期。Redis 的核心守护任务 serverCron 每次返回 1000/server.hz，在默认配置（hz = 10）下维持着每秒 10 次的稳定调度节拍。

删除操作则采用了惰性清理策略。主动注销或者一次性事件执行完毕时，事件对象的 id 首先被标记为 AE_DELETED_EVENT_ID。只有在主循环安全遍历且节点引用计数 refcount 归零时，链表才会将其真正脱链并释放内存。这种机制有效规避了在回调执行期间修改链表拓扑导致的悬空指针与迭代器失效。

5. 平台抽象层

I/O 多路复用的平台适配完全在编译期敲定。在 ae.c 源码顶部，预处理器通过条件编译指令检测当前系统的系统调用支持，从高到低挑选性能最优的底层后端：

#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

各个实现文件的选型优先级由高到低依次为：Solaris 的 evport、Linux 的 epoll、macOS 与 BSD 系列的 kqueue，最后以 POSIX 标准普遍支持的 select 作为全平台兜底。每一个平台适配文件都遵循约定，分别实现一组签名完全一致的内部静态函数：

函数	职责
`aeApiCreate`	初始化底层数据结构（如 `epoll_create`）
`aeApiResize`	调整容量
`aeApiFree`	释放资源
`aeApiAddEvent`	注册 fd 监听（如 `epoll_ctl`）
`aeApiDelEvent`	取消监听
`aeApiPoll`	阻塞等待事件（如 `epoll_wait`）
`aeApiName`	返回名字（`"epoll"`、`"kqueue"` 等）

用 #include .c 实现编译期多态

ae 在架构上没有借助虚函数指针表（vtable）或策略结构体来完成运行时多态分发，而是直接利用预处理器 #include 对应的 .c 源码文件。被包含的代码直接融入 ae.c，使得所有平台相关的 static 函数与事件循环主逻辑共处同一个编译翻译单元。这一设计带来了三项直接优势：

零间接寻址开销：所有底层调用均为直接调用，现代编译器能够直接进行跨函数内联优化，消除了虚表跳转的寻址延迟。
约定优于配置：无需定义繁杂的抽象基类或结构体接口，仅依赖同名函数的隐式签名约定即可完成平台替换。
目标体积精简：产出的可执行文件仅包含目标系统所需的那一种系统调用逻辑，绝不打包无关平台的冗余代码。

以 Linux 下的 epoll 实现为例，aeApiPoll 的内部处理非常直观：

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 + 999)/1000) : -1);
    if (retval > 0) {
        numevents = retval;
        for (int 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;
            eventLoop->fired[j].fd = e->data.fd;
            eventLoop->fired[j].mask = mask;
        }
    }
    return numevents;
}

该函数的核心职责是将内核返回的 EPOLLIN 与 EPOLLOUT 事件，无损转换为 ae 统一的 AE_READABLE 与 AE_WRITABLE 掩码，并整齐地排入 eventLoop->fired 数组供上层分发。

值得留意的细节是错误处理：内核派发的 EPOLLERR 与 EPOLLHUP（连接挂断）被同时映射为读与写双重就绪。这样一来，无论上层在当前文件描述符上仅注册了读回调还是仅注册了写回调，都能够立刻被唤醒并感知到异常状态，进而执行套接字关闭与清理流程，避免连接在错误状态下无声挂死。

6. beforesleep / aftersleep 钩子

void aeSetBeforeSleepProc(aeEventLoop *eventLoop, aeBeforeSleepProc *beforesleep) {
    eventLoop->beforesleep = beforesleep;
}

void aeSetAfterSleepProc(aeEventLoop *eventLoop, aeBeforeSleepProc *aftersleep) {
    eventLoop->aftersleep = aftersleep;
}

在 aeProcessEvents 的主执行流中，beforesleep 安排在调用 aeApiPoll 阻塞等待之前执行，而 aftersleep 则挂接在底层轮询返回之后立即触发。

这两个生命周期钩子是 Redis 单线程吞吐与低延迟设计的关键枢纽。在调用多路复用系统调用沉睡之前，系统通过 beforesleep 集中处理大量缓冲逻辑：主动遍历客户端就绪列表，将待发送的数据批量推入网络套接字（避免每次写入都陷入一次系统调用）；将 AOF 缓冲区的数据集中刷入磁盘；处理被阻塞键唤醒的客户端请求；同步集群内部的心跳信息等。而在系统调用醒来后，aftersleep 能够第一时间刷新当前的系统时间缓存与模块延迟统计，让后续的事件分发逻辑获得精度最高的运行环境。

7. Redis 网络层如何接入 ae

ae 自身仅提供抽象的文件描述符分发与时间事件轮询，并不直接承载任何网络协议解析与套接字管理细节。Redis 巧妙地将整套网络状态机架设在 ae 暴露的文件事件之上：

8. 与通用事件库的对比

ae 的一切权衡都完全围绕 Redis 自身的特质展开：文件事件直接采用以描述符数值为下标的紧凑数组，时间事件仅维护简单的无序单链表，底层系统调用多路复用在编译期通过宏替换一锤定音。它没有附带异步 DNS 解析、跨平台工作线程池或异步文件系统 I/O 等重量级组件，而是将事件引擎的体量压缩到了千行以内。

9. 习题

9.1 巩固

#50514

⚡3⏳2

(1) [1.1] 不运行程序，回答 aeCreateEventLoop(64) 与 aeCreateEventLoop(4096) 之后 setsize、nevents 各是多少，events 与 fired 两个数组各分配了多少个元素。再答：setsize 为 64 时调用 aeCreateFileEvent(loop, 99, AE_READABLE, proc, NULL) 返回什么，errno 是多少？写下你的预测，再用正文的源码核对，或写十行程序打印 aeGetSetSize 与返回值验证。

(2) [1.2] 时间事件链表里同时存在三个事件：id 为 3、when 为 900，id 为 1、when 为 1200，id 为 2、when 为 700。当前单调时钟是 500 微秒。usUntilEarliestTimer 返回多少？如果 id 为 2 的事件已经被 aeDeleteTimeEvent 标记删除，返回值又变成多少？两次都不看代码，按函数的循环写出推导过程。

(3) [3.2] aeFileEvent 里 rfileProc 与 wfileProc 可以指向同一个函数。分发循环里哪一行判断负责避免同一个回调被调用两次？如果把这一行去掉，同一个 fd 上的这两个回调会各被调用几次？用 socketpair 造一个同时可读可写的 fd 验证。

#50515

⚡3⏳3

写一个程序，只注册一个 20 毫秒后触发的周期定时器（回调返回 20）和一个 beforesleep、一个 aftersleep，三处各打印一行带时刻的标记。调用一次 aeProcessEvents(loop, AE_ALL_EVENTS|AE_CALL_BEFORE_SLEEP|AE_CALL_AFTER_SLEEP)，记录三行输出的先后，回答：

两个钩子各自在 aeApiPoll 的哪一侧执行，一次调用里各执行几次？
定时器回调相对两个钩子的位置说明了 aeProcessEvents 的四步流程里时间事件排在第几步？
把 AE_DONT_WAIT 加到 flags 里、循环里没有任何文件和定时器时，函数立即返回还是阻塞？用 clock_gettime(CLOCK_MONOTONIC) 前后计时给出毫秒数。

验收：三行带时刻的输出、三个问题的回答、空循环计时结果。

9.2 编程

#50516

⚡4⏳4

把 ae 从 Redis 源码树里摘出来，做成一个不依赖 redis-server 的独立程序。

从 8.10.2 的 src/ 复制 ae.c、ae.h、ae_epoll.c、ae_select.c、anet.h（只要头文件，ae.c 会 #include 它）、zmalloc.c、zmalloc.h、monotonic.c、monotonic.h、redisassert.c、redisassert.h、config.h、fmacros.h、atomicvar.h、solarisfixes.h 到一个空目录。
先去掉 redisassert.c 编译一次，看链接器报出哪两个未定义符号，再到 ae.c、monotonic.c、zmalloc.c 里找出引用它们的 assert 与 panic 各在哪几行。补回 redisassert.c 之后还剩一个：epoll 后端在 aeApiCreate 里调用的 anetCloexec。写一个三行的 shim.c，用 fcntl 的 F_GETFD、F_SETFD 实现它（只设 FD_CLOEXEC），不要复制整个 anet.c。最后列出全部缺失符号、来源文件与提供者。
写主程序：创建 setsize 为 32 的循环，打印 aeGetApiName 与 aeGetSetSize；在标准输入上注册 AE_READABLE，读一行原样打印；再注册一个 100 毫秒的一次性定时器，触发后调用 aeStop。用 printf 'hello\n' | ./aedemo 运行。
再写一个只用定时器的版本：三个周期定时器分别每 30、70、110 毫秒触发一次，各触发 3 次后返回 AE_NOMORE，打印每次触发的 id 与到达时刻。全部结束后程序退出。记录三个定时器的触发次数之和是 9。

验收：编译命令与缺失符号清单、两个程序的实际输出、第 4 步的 9 行触发记录。

#50517

⚡4⏳3

用 socketpair(AF_UNIX, SOCK_STREAM, 0, sv) 造一对已连接的 fd，在一个 ae 循环里复现文件事件的分发顺序。

要求：

给 sv[0] 注册可读与可写，读回调用一个函数、写回调用另一个函数，两者各往一个数组里记一个编号。写一个字节到 sv[1]，调用一次 aeProcessEvents(loop, AE_ALL_EVENTS|AE_DONT_WAIT)，打印回调顺序。应当是先读后写。
用 aeDeleteFileEvent(loop, sv[0], AE_WRITABLE) 摘掉写事件，再打印 aeGetFileEvents 的返回值，确认它同时清掉了 AE_BARRIER。然后用 AE_WRITABLE|AE_BARRIER 重新注册，再写一个字节、再跑一轮，顺序应当反过来。
把读写两个回调改成同一个函数，只注册一次 AE_READABLE|AE_WRITABLE，在回调里读走那个字节，确认该函数一轮只被调用一次。
让 aeGetFileEvents 查询一个没有注册过的 fd 和编号大于 setsize 的 fd，各返回什么。

验收：四步各一组输出，并对第 1、2 步的顺序差异给出一句解释，引用分发循环里的 invert 标志。

#50518

⚡5⏳4

usUntilEarliestTimer 是 O(N) 扫描，processTimeEvents 也是 O(N)。把时间事件链表改成按 when 升序的有序链表，让前者变成 O(1)、后者只需扫描前缀。

要求：

保留 aeCreateTimeEvent、aeDeleteTimeEvent、aeTimeEvent 的公开接口与惰性删除语义（AE_DELETED_EVENT_ID），processTimeEvents 里「不处理本轮新建的事件」那条 id > maxId 判断也要保留。插入时按 when 找位置，复杂度 O(N)。
写测试：随机生成 10000 个定时器，延迟在 1～10000 毫秒之间，另加 500 个随机 aeDeleteTimeEvent。两种实现各跑一遍，记录每个定时器回调被调用的 id 顺序，两个序列必须逐项相同。
在 usUntilEarliestTimer 里加一个计数器统计循环次数，对同样的 10000 个定时器调用 1000 次 aeProcessEvents(loop, AE_ALL_EVENTS|AE_DONT_WAIT)，比较原实现与有序链表的计数。再测两种实现的建表耗时，说明各自适合什么规模。

验收：两个 id 序列一致、两组循环计数（原实现应当接近 1000 万次量级，有序链表为 0）、建表耗时对照。

#50519

⚡5⏳4

正文里定时器回调返回正整数后，下一次 when 是「现在加间隔」。写程序把这个行为量出来。

注册一个返回 100 的周期定时器，回调里用 nanosleep 睡 10 毫秒，记录第 1 次与第 100 次触发的时刻差。预测这个差值，再实测。参考量级：本机按这套办法跑出的 99 个间隔平均是 110 毫秒左右，不是 100。
把 ae.c 里 te->when = now + (monotime)retval * 1000; 改成按固定节拍推进的写法（记住上一次的 when，每次加 retval * 1000，并处理回调耗时超过一个周期的情况），重跑同一实验。
两种实现的第 100 次触发时刻差、累计漂移各是多少？serverCron 用 1000/hz 作为返回值，按第一种写法，每秒 10 次的调度在命令处理很忙时会变成多少次？

验收：两种实现的时刻差与漂移数据、第 3 个问题的一句结论。

#50520

⚡6⏳5

参照 ae_epoll.c 与 ae_select.c，写一个新的后端 ae_poll.c，用 poll(2) 实现同样的七个 static 函数：aeApiCreate、aeApiResize、aeApiFree、aeApiAddEvent、aeApiDelEvent、aeApiPoll、aeApiName。

要求：

循环里维护一个 struct pollfd 数组，下标就是 fd；aeApiAddEvent 与 aeApiDelEvent 都要维护 events 与 revents，未注册的槽位 fd 置 -1。
aeApiPoll 返回就绪个数，并把 POLLIN、POLLOUT、POLLERR、POLLHUP 翻译成 AE_READABLE、AE_WRITABLE，翻译规则与 epoll 后端一致。
用编译开关选择后端：复制一份 config.h 并注释掉 #define HAVE_EPOLL 1，确认 ae.c 走 #include "ae_select.c" 的分支；再把那一行换成 #include "ae_poll.c" 的入口，对比三种后端跑同一个演示程序（socketpair 加一个 50 毫秒定时器）时的输出是否一致。
列一张表：三种后端在接口形态、单次等待的系统开销、fd 数量上限、setsize 如何限制注册上的差异。再单独回答 select 后端的一个现象：直接 aeCreateEventLoop(2000) 会成功，而对一个 setsize 为 64 的循环调用 aeResizeSetSize(loop, 2000) 返回 -1。两处分别做了什么检查？把两次调用的返回值与之后的 aeGetSetSize 都打印出来。select 后端此时并没有阻止你注册 fd 编号大于等于 FD_SETSIZE 的描述符，aeApiAddEvent 里的 FD_SET 会发生什么？

验收：ae_poll.c 全文、三种后端的同一演示输出、差异表，以及 select 后端在 setsize 为 2000 时的实际返回。

9.3 综合与挑战

#50521

⚡7⏳5选做

做一个实验，验证 AE_BARRIER 存在的理由：回复必须在落盘之后才发出。

要求：

写一个单线程服务器，用 ae 监听本机端口。每个连接注册可读回调，收到一行就追加到 data.log 并 fsync，然后回复 OK。回复先放进连接自己的输出缓冲区。
第一版在可读回调里直接发回复；第二版改成「可读回调只登记可写事件，可写回调负责发送」，并按 Redis 的做法在写事件上带 AE_BARRIER。
客户端一次连发 1000 行，不读回复，直到发送阻塞。用 strace -f -e trace=read,write,fsync,epoll_wait 抓两版的系统调用序列，统计每一版里「同一轮 epoll_wait 之后 read 与 fsync 与 write 的先后」。
回答：AE_BARRIER 翻转的是读回调与写回调的顺序，为什么这个顺序能让「先 fsync 再回复」成立？如果写回调里只发了一半就因为发送缓冲区满而返回，剩下的部分与下一批请求之间是什么关系？

验收：两版各一段 strace 摘录、两版里 fsync 与 write 的先后对照、两问的回答。AE_BARRIER 只影响同一轮里两种回调的顺序，不要把它说成能保证跨轮次的顺序。

3.10 求值

4.3 选择语句

5.5 跳转语句

10.2 结构体

14.3 程序支持

15.1 声明

16.2 对象表示

18.2 标准库中的宏

5.3 源码阅读：Redis ae 事件循环 ​

1. 核心数据结构 ​

文件事件：fd + mask + 回调

时间事件：链表 + 单调时钟

2. 主循环：aeMain 与 aeProcessEvents ​

aeProcessEvents 的四步流程

3. 文件事件的注册与分发 ​

3.1 注册 ​

3.2 分发与 AE_BARRIER ​

4. 时间事件的管理 ​

4.1 创建 ​

4.2 处理与重调度 ​

5. 平台抽象层 ​