ae事件循环
ae 事件循环 源码分析(单线程 Reactor + IO 线程分担读写)
ae 是 Redis 自己写的极简事件库,解决的问题:一个线程如何同时伺候几万个客户端连接,还要顺带执行定时任务。
核心思路是 Reactor 模式——把「等 IO」交给内核多路复用(epoll),把「处理事件」留在单线程里顺序执行,等待超时精确设成距离最近定时器的时间,让一个 epoll_wait 同时兼顾网络事件和定时器。
// 基于本地 Redis 仓 (unstable), src/ae.h
#define AE_NONE 0 /* No events registered. */
#define AE_READABLE 1 /* Fire when descriptor is readable. */
#define AE_WRITABLE 2 /* Fire when descriptor is writable. */
#define AE_BARRIER 4 // 与 WRITABLE 联用：本轮已触发过读事件就先写后读反转顺序，用于"先 fsync 再回包"
/* File event structure */
typedef struct aeFileEvent {
int mask; /* one of AE_(READABLE|WRITABLE|BARRIER) */
aeFileProc *rfileProc; // 读回调，Redis 里注册的是 readQueryFromClient
aeFileProc *wfileProc; // 写回调，如 sendReplyToClient
void *clientData; // 一般挂 connection/client 指针
} aeFileEvent;
/* Time event structure */
typedef struct aeTimeEvent {
long long id; /* time event identifier. */
monotime when; // 绝对触发时刻（单调时钟微秒），不受系统调时间影响
aeTimeProc *timeProc; // Redis 里几乎只有一个：serverCron
aeEventFinalizerProc *finalizerProc;
void *clientData;
struct aeTimeEvent *prev; // 无序双向链表——Redis 定时器极少，O(N) 扫也无所谓
struct aeTimeEvent *next;
int refcount; /* refcount to prevent timer events from being
* freed in recursive time event calls. */
} aeTimeEvent;
/* State of an event based program */
typedef struct aeEventLoop {
int maxfd; /* highest file descriptor currently registered */
int setsize; /* max number of file descriptors tracked */
long long timeEventNextId;
int nevents; /* Size of Registered events */
aeFileEvent *events; /* Registered events */ // 关键设计：fd 直接当数组下标，O(1) 找回调
aeFiredEvent *fired; /* Fired events */ // epoll_wait 返回的就绪事件抄到这里
aeTimeEvent *timeEventHead;
int stop;
void *apidata; /* This is used for polling API specific data */ // epoll 版里是 aeApiState
aeBeforeSleepProc *beforesleep; // 每轮睡前钩子，Redis 把大量收尾工作塞在这
aeBeforeSleepProc *aftersleep;
int flags;
void *privdata[2];
} aeEventLoop;// 基于本地 Redis 仓 (unstable), src/ae.c
void aeMain(aeEventLoop *eventLoop) {
eventLoop->stop = 0;
while (!eventLoop->stop) { // 整个 Redis 主线程的一生就是这个 while
aeProcessEvents(eventLoop, AE_ALL_EVENTS|
AE_CALL_BEFORE_SLEEP|
AE_CALL_AFTER_SLEEP);
}
}
int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,
aeFileProc *proc, void *clientData)
{
if (fd >= eventLoop->setsize) {
errno = ERANGE;
return AE_ERR;
}
// ... fd 超过 nevents 时对 events/fired 数组翻倍扩容，略
aeFileEvent *fe = &eventLoop->events[fd]; // fd 即下标，无需哈希表
if (aeApiAddEvent(eventLoop, fd, mask) == -1) // 落到 epoll_ctl
return AE_ERR;
fe->mask |= mask; // 或运算：同一 fd 可同时挂读+写
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;
}
/* How many microseconds until the first timer should fire.
* If there are no timers, -1 is returned. */
// ...
static int64_t usUntilEarliestTimer(aeEventLoop *eventLoop) {
aeTimeEvent *te = eventLoop->timeEventHead;
if (te == NULL) return -1;
aeTimeEvent *earliest = NULL;
while (te) { // O(N) 遍历找最早的定时器，N 极小所以不排序
if ((!earliest || te->when < earliest->when) && te->id != AE_DELETED_EVENT_ID)
earliest = te;
te = te->next;
}
monotime now = getMonotonicUs();
return (now >= earliest->when) ? 0 : earliest->when - now; // 已过期返回 0 = 不睡
}
int aeProcessEvents(aeEventLoop *eventLoop, int flags)
{
int processed = 0, numevents;
/* Nothing to do? return ASAP */
if (!(flags & AE_TIME_EVENTS) && !(flags & AE_FILE_EVENTS)) return 0;
// 注意条件：即使一个 fd 都没有，只要有时间事件也要走 aeApiPoll——借它睡到定时器到期
if (eventLoop->maxfd != -1 ||
((flags & AE_TIME_EVENTS) && !(flags & AE_DONT_WAIT))) {
int j;
struct timeval tv, *tvp = NULL; /* NULL means infinite wait. */
int64_t usUntilTimer;
if (eventLoop->beforesleep != NULL && (flags & AE_CALL_BEFORE_SLEEP))
eventLoop->beforesleep(eventLoop); // 睡前钩子：回写客户端、刷 AOF 等
// ... 注释略：beforesleep 内可能改 eventLoop->flags，所以在其后再判断
if ((flags & AE_DONT_WAIT) || (eventLoop->flags & AE_DONT_WAIT)) {
tv.tv_sec = tv.tv_usec = 0; // 还有活没干完，poll 一下立刻返回
tvp = &tv;
} else if (flags & AE_TIME_EVENTS) {
usUntilTimer = usUntilEarliestTimer(eventLoop); // 第一步：算最近定时器还有多久
if (usUntilTimer >= 0) {
tv.tv_sec = usUntilTimer / 1000000;
tv.tv_usec = usUntilTimer % 1000000;
tvp = &tv; // 把它作为 epoll_wait 的超时——定时器不需要独立线程
}
}
/* Call the multiplexing API, will return only on timeout or when
* some event fires. */
numevents = aeApiPoll(eventLoop, tvp); // 第二步：阻塞在这里，唯一的"睡觉"点
/* Don't process file events if not requested. */
if (!(flags & AE_FILE_EVENTS)) {
numevents = 0;
}
/* After sleep callback. */
if (eventLoop->aftersleep != NULL && flags & AE_CALL_AFTER_SLEEP)
eventLoop->aftersleep(eventLoop);
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; /* Number of events fired for current fd. */
// ... 注释略：常规顺序是先读后写——读完命令马上能把回复写出去，省一轮循环
int invert = fe->mask & AE_BARRIER; // BARRIER 反转成先写后读
// fe->mask & mask 再查一次：前面的回调可能已经把这个事件删了
if (!invert && fe->mask & mask & AE_READABLE) {
fe->rfileProc(eventLoop,fd,fe->clientData,mask);
fired++;
fe = &eventLoop->events[fd]; /* Refresh in case of resize. */ // 回调里可能 realloc 了 events 数组
}
/* Fire the writable event. */
if (fe->mask & mask & AE_WRITABLE) {
if (!fired || fe->wfileProc != fe->rfileProc) { // 读写回调是同一个函数时不重复调
fe->wfileProc(eventLoop,fd,fe->clientData,mask);
fired++;
}
}
/* If we have to invert the call, fire the readable event now
* after the writable one. */
if (invert) {
fe = &eventLoop->events[fd]; /* Refresh in case of resize. */
if ((fe->mask & mask & AE_READABLE) &&
(!fired || fe->wfileProc != fe->rfileProc))
{
fe->rfileProc(eventLoop,fd,fe->clientData,mask);
fired++;
}
}
processed++;
}
}
/* Check time events */
if (flags & AE_TIME_EVENTS)
processed += processTimeEvents(eventLoop); // 第四步：最后才处理时间事件
return processed; /* return the number of processed file/time events */
}// 基于本地 Redis 仓 (unstable), src/ae_epoll.c —— Linux 下编译期选中的多路复用实现
typedef struct aeApiState {
int epfd; // epoll 实例的 fd
struct epoll_event *events; // epoll_wait 的输出缓冲区
} aeApiState;
static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
aeApiState *state = eventLoop->apidata;
struct epoll_event ee = {0}; /* avoid valgrind warning */
/* If the fd was already monitored for some event, we need a MOD
* operation. Otherwise we need an ADD operation. */
int op = eventLoop->events[fd].mask == AE_NONE ?
EPOLL_CTL_ADD : EPOLL_CTL_MOD; // 靠 ae 自己记的 mask 区分 ADD/MOD，不用试错
ee.events = 0;
mask |= eventLoop->events[fd].mask; /* Merge old events */ // 新旧事件合并，MOD 是全量覆盖语义
if (mask & AE_READABLE) ee.events |= EPOLLIN;
if (mask & AE_WRITABLE) ee.events |= EPOLLOUT;
ee.data.fd = fd;
if (epoll_ctl(state->epfd,op,fd,&ee) == -1) return -1;
return 0;
}
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); // 微秒向上取整成毫秒；tvp==NULL 则无限等
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; // 出错转成读写事件，让回调里的 read/write 拿到错误码
if (e->events & EPOLLHUP) mask |= AE_WRITABLE|AE_READABLE;
eventLoop->fired[j].fd = e->data.fd; // 翻译成平台无关的 fired 数组
eventLoop->fired[j].mask = mask;
}
} else if (retval == -1 && errno != EINTR) {
panic("aeApiPoll: epoll_wait, %s", strerror(errno));
}
return numevents;
}原理串讲
一次完整的循环:main 里调 aeMain,死循环调 aeProcessEvents。
每轮先执行 beforesleep 钩子(Redis 注册的是 server.c 的 beforeSleep),然后 usUntilEarliestTimer 遍历定时器链表算出最近一个定时器还有多少微秒到期,把这个值填进 timeval 传给 aeApiPoll,最终成为 epoll_wait 的超时参数。
于是线程要么被网络事件唤醒,要么恰好睡到定时器该触发的时刻醒来——这是第一个「为什么」:定时器不需要额外线程或信号,把”最近的闹钟”折算成 epoll 超时,一个阻塞点就同时管住了 IO 和时间,代价只是定时器精度受限于本轮循环长度,所以 serverCron 的设计也是”到点了做一小片活”而不是精确调度。
醒来后先遍历 fired 数组处理文件事件:从 events[fd] 取出回调,默认先 rfileProc 后 wfileProc——先读是因为读完命令、执行完,回复往往已经躺在输出缓冲区里,紧接着的写事件就能立刻把它发出去;文件事件全部处理完才轮到 processTimeEvents 扫链表执行到期的 serverCron。
beforeSleep(server.c)是理解 Redis 的关键钩子,它承包了”命令执行完到真正入睡之间”的所有收尾:快速过期扫描 activeExpireCycle(ACTIVE_EXPIRE_CYCLE_FAST)、唤醒解除阻塞的客户端 blockedBeforeSleep、把 AOF 缓冲刷盘 flushAppendOnlyFile,以及最重要的 handleClientsWithPendingWrites——把本轮所有命令产生的回复统一回写。
这是第二个「为什么」:命令执行时只把回复追加到 client 的输出缓冲区并把 client 挂进 pending 队列,不立即 write。
这样同一个客户端在一轮里发的多条流水线命令,回复会攒起来一次 write 系统调用发走,syscall 次数从 N 降到 1;只有一次没写完的,才注册 AE_WRITABLE 事件交给下一轮。
AE_BARRIER 服务于同一个钩子:appendfsync=always 时必须保证 beforeSleep 里先 fsync 再回包,所以要禁止”读到命令后顺手把上一轮回复写出去”的默认顺序。
代码块收起展开
单线程为什么快:没有锁、没有上下文切换、没有并发数据结构,所有对键空间的访问天然串行,这让 dict/ziplist 这些结构可以做得极其紧凑;而瓶颈本来就不在 CPU 计算——大多数命令是 O(1)/O(logN) 的内存操作,真正的开销在网络 IO 的 read/write 和协议解析上。
所以 unstable 版的多线程方案(src/iothread.c)也只动 IO 不动执行:`initThreadedIO` 为每个 IO 线程各建一个独立的 `aeEventLoop`,`IOThreadMain` 里同样跑 `aeMain`,客户端连接被 `assignClientToIOThread` 绑定到某个 IO 线程后,读事件、协议解析、回复写出都在那个线程的事件循环里完成;但一旦解析出完整命令,client 会被挂进带互斥锁的 `mainThreadPendingClients` 链表并通过 eventNotifier 唤醒主线程,主线程在 `processClientsFromIOThread` 里逐个取出、置 `running_tid` 为主线程后调 `processPendingCommandAndInputBuffer` 执行命令,执行完再把 client 交还 IO 线程去写回复。命令执行始终单线程,键空间依旧无锁,多线程只是把最贵的 read/write/解析摊出去——两边通过”同一时刻一个 client 只属于一个线程”来避免加锁访问 client 数据。
设计取舍
- 定时器用无序链表,找最近定时器 O(N):源码注释自己承认可以换跳表,但 Redis 定时器就一个 serverCron 量级,不值得。
events数组用 fd 直接当下标:O(1) 且无哈希开销,代价是数组大小要按 maxclients 预留,fd 稀疏时浪费内存——服务器场景 fd 密集,划算。- 先读后写不是随便定的:读→执行→写在同一轮完成,省掉一整轮事件循环的延迟;
AE_BARRIER是这个优化的逃生门,不是常态。 - 常见误区:「Redis 6 之后是多线程了」——IO 线程只做 read/解析/write,命令执行和键空间访问从来都在主线程,事务和 Lua 的原子性保证不变。
epoll_wait超时向上取整((tv_usec + 999)/1000):宁可多睡不提前醒,提前醒会发现定时器没到期又空转一圈。