Redis事件驱动模型源码深度拆解

一、线上故障：Redis连接风暴导致服务雪崩

2024年3月，我们一个日活500万的电商业务出现诡异故障：Redis CPU飙升到95%，连接数从2000暴涨到8000，QPS从12万跌到3万。排查发现是上游服务突发重连风暴，大量TIME_WAIT连接堆积，Redis事件循环处理不过来了。

这个事故让我决定彻底搞懂Redis事件驱动模型——为什么连接数一多就扛不住？select/poll/epoll到底差在哪？

二、方案对比：select vs poll vs epoll

Redis支持三种事件驱动后端：select（默认）、poll、epoll（Linux推荐）。我们压测对比一下。

2.1 测试环境

• 服务器：4核8G CentOS 7.9，内核3.10.0
• Redis版本：7.2.4（源码编译）
• 压测工具：redis-benchmark 7.2.4
• 连接数：1000个并发客户端

2.2 压测结果

epoll比select QPS提升39%，CPU降低20%。原因在于select的O(n)轮询 vs epoll的O(1)事件通知。

三、源码分析：aeEventLoop核心结构

Redis事件驱动代码在 src/ae.c 和 src/ae.h，版本7.2.4。核心数据结构是 aeEventLoop。

// ae.h 第55-80行
typedef struct aeEventLoop {
    int maxfd;                   // 当前最大文件描述符
    int setsize;                 // 最大事件数（默认1024）
    long long timeEventNextId;   // 定时器ID自增
    time_t lastTime;             // 上次处理时间
    aeFileEvent *events;         // 文件事件数组
    aeFiredEvent *fired;         // 就绪事件数组
    aeTimeEvent *timeEventHead;  // 定时器链表头
    int stop;                    // 停止标志
    void *apidata;               // 多路复用API数据（epoll/poll/select）
    aeBeforeSleepProc *beforesleep; // 睡眠前回调
    aeBeforeSleepProc *aftersleep;  // 睡眠后回调
} aeEventLoop;

关键字段：apidata 指向具体后端的数据结构（如epoll的 aeApiState），events 和 fired 是事件数组，timeEventHead 是定时器链表。

四、事件循环主流程：aeMain

入口函数 aeMain 在 ae.c 第500行。

// ae.c 第500-515行
void aeMain(aeEventLoop *eventLoop) {
    eventLoop->stop = 0;
    while (!eventLoop->stop) {
        // 睡眠前回调（如处理集群消息）
        if (eventLoop->beforesleep != NULL)
            eventLoop->beforesleep(eventLoop);
        
        // 核心：等待事件就绪
        aeProcessEvents(eventLoop, AE_ALL_EVENTS | AE_CALL_BEFORE_SLEEP | AE_CALL_AFTER_SLEEP);
    }
}

循环体核心是 aeProcessEvents，它处理文件事件和定时器事件。

五、文件事件处理：aeProcessEvents

这个函数在 ae.c 第380行，逻辑清晰：先计算最近定时器时间，然后调用多路复用API等待事件，最后处理就绪事件。

// ae.c 第380-450行（简化）
int aeProcessEvents(aeEventLoop *eventLoop, int flags) {
    int processed = 0;
    struct timeval tv, *tvp;
    
    // 1. 计算定时器最近触发时间
    if (flags & AE_TIME_EVENTS) {
        long long shortest = aeSearchNearestTimer(eventLoop);
        if (shortest >= 0) {
            tv.tv_sec = shortest / 1000;
            tv.tv_usec = (shortest % 1000) * 1000;
            tvp = &tv;
        } else {
            tvp = NULL; // 无定时器则阻塞等待
        }
    }
    
    // 2. 调用多路复用API等待事件
    if (flags & AE_FILE_EVENTS) {
        int numevents = aeApiPoll(eventLoop, tvp);
        for (int j = 0; j < numevents; j++) {
            aeFileEvent *fe = &eventLoop->events[eventLoop->fired[j].fd];
            int mask = eventLoop->fired[j].mask;
            
            // 3. 处理读事件
            if (fe->rfileProc && (mask & AE_READABLE))
                fe->rfileProc(eventLoop, fd, fe->clientData, mask);
            // 4. 处理写事件
            if (fe->wfileProc && (mask & AE_WRITABLE))
                fe->wfileProc(eventLoop, fd, fe->clientData, mask);
        }
    }
    
    // 5. 处理定时器事件
    if (flags & AE_TIME_EVENTS)
        processed += processTimeEvents(eventLoop);
    
    return processed;
}

注意：aeApiPoll 是平台相关的，Linux下调用 epoll_wait。

六、epoll封装实现：ae_epoll.c

Redis对epoll的封装在 src/ae_epoll.c，核心三个函数：aeApiCreate、aeApiAddEvent、aeApiPoll。

6.1 aeApiCreate：创建epoll实例

// ae_epoll.c 第30-50行
static int aeApiCreate(aeEventLoop *eventLoop) {
    aeApiState *state = zmalloc(sizeof(aeApiState));
    if (!state) return -1;
    
    // 创建epoll实例，size参数被忽略（Linux 2.6.8+）
    state->epfd = epoll_create(1024);
    if (state->epfd == -1) {
        zfree(state);
        return -1;
    }
    
    // 分配就绪事件数组
    state->events = zmalloc(sizeof(struct epoll_event) * eventLoop->setsize);
    if (!state->events) {
        close(state->epfd);
        zfree(state);
        return -1;
    }
    
    eventLoop->apidata = state;
    return 0;
}

6.2 aeApiAddEvent：注册事件

// ae_epoll.c 第60-80行
static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
    aeApiState *state = eventLoop->apidata;
    struct epoll_event ee = {0};
    
    // 如果已注册，用EPOLL_CTL_MOD，否则用EPOLL_CTL_ADD
    int op = eventLoop->events[fd].mask == AE_NONE ?
             EPOLL_CTL_ADD : EPOLL_CTL_MOD;
    
    ee.events = 0;
    if (mask & AE_READABLE) ee.events |= EPOLLIN;
    if (mask & AE_WRITABLE) ee.events |= EPOLLOUT;
    ee.data.fd = fd;
    
    return epoll_ctl(state->epfd, op, fd, &ee);
}

6.3 aeApiPoll：等待事件

// ae_epoll.c 第90-120行
static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
    aeApiState *state = eventLoop->apidata;
    int retval, numevents = 0;
    
    // 调用epoll_wait，超时时间由定时器决定
    retval = epoll_wait(state->epfd, state->events, eventLoop->setsize,
                        tvp ? (tvp->tv_sec * 1000 + tvp->tv_usec / 1000) : -1);
    
    if (retval > 0) {
        numevents = retval;
        for (int i = 0; i < numevents; i++) {
            int mask = 0;
            struct epoll_event *e = &state->events[i];
            
            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[i].fd = e->data.fd;
            eventLoop->fired[i].mask = mask;
        }
    }
    
    return numevents;
}

关键点：epoll_wait 返回就绪事件数组，Redis直接遍历处理，复杂度O(1)（相对于select的O(n)扫描）。

七、定时器实现：时间事件链表

Redis定时器用单向链表管理，每个节点是 aeTimeEvent。

// ae.h 第85行
typedef struct aeTimeEvent {
    long long id;                 // 定时器ID
    long when_sec;                // 触发时间（秒）
    long when_ms;                 // 触发时间（毫秒）
    aeTimeProc *timeProc;         // 回调函数
    aeEventFinalizerProc *finalizerProc; // 清理函数
    void *clientData;             // 用户数据
    struct aeTimeEvent *next;     // 链表指针
} aeTimeEvent;

处理函数 processTimeEvents 遍历链表，执行到期定时器。

// ae.c 第300-350行（简化）
static int processTimeEvents(aeEventLoop *eventLoop) {
    int processed = 0;
    aeTimeEvent *te, *prev = NULL;
    long long maxId = eventLoop->timeEventNextId - 1;
    
    te = eventLoop->timeEventHead;
    while (te) {
        // 检查是否到期
        if (te->when_sec < now_sec || 
            (te->when_sec == now_sec && te->when_ms <= now_ms)) {
            int retval = te->timeProc(eventLoop, te->id, te->clientData);
            processed++;
            
            // 如果返回AE_NOMORE则删除，否则更新下次触发时间
            if (retval != AE_NOMORE) {
                te->when_sec = now_sec + retval / 1000;
                te->when_ms = now_ms + retval % 1000;
            } else {
                // 删除定时器
                if (prev) prev->next = te->next;
                else eventLoop->timeEventHead = te->next;
                zfree(te);
            }
        }
        prev = te;
        te = te->next;
    }
    return processed;
}

注意：Redis定时器是单线程处理的，所以链表操作不需要加锁。但这也意味着定时器回调不能阻塞太久。

八、完整事件循环示例：模拟Redis处理客户端

下面是一个简化版的事件循环，演示如何用Redis的ae库处理TCP连接。

// 完整可运行示例：simple_redis_ae.c
#include "ae.h"
#include 
#include 
#include 
#include 
#include 
#include 

#define PORT 6379
#define MAX_CLIENTS 1024

// 读事件回调：处理客户端请求
void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
    char buf[1024];
    int nread = read(fd, buf, sizeof(buf) - 1);
    if (nread <= 0) {
        // 客户端断开
        close(fd);
        aeDeleteFileEvent(el, fd, AE_READABLE);
        return;
    }
    buf[nread] = '\0';
    printf("收到: %s", buf);
    
    // 简单回复PONG
    const char *reply = "+PONG\r\n";
    write(fd, reply, strlen(reply));
}

// 接受新连接
void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
    int client_fd;
    struct sockaddr_in client_addr;
    socklen_t client_len = sizeof(client_addr);
    
    client_fd = accept(fd, (struct sockaddr *)&client_addr, &client_len);
    if (client_fd == -1) return;
    
    // 注册客户端读事件
    aeCreateFileEvent(el, client_fd, AE_READABLE, readQueryFromClient, NULL);
    printf("新客户端连接: %d\n", client_fd);
}

int main() {
    // 创建事件循环
    aeEventLoop *el = aeCreateEventLoop(MAX_CLIENTS);
    if (!el) {
        fprintf(stderr, "创建事件循环失败\n");
        return 1;
    }
    
    // 创建TCP监听socket
    int listen_fd = socket(AF_INET, SOCK_STREAM, 0);
    int opt = 1;
    setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
    
    struct sockaddr_in server_addr;
    memset(&server_addr, 0, sizeof(server_addr));
    server_addr.sin_family = AF_INET;
    server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
    server_addr.sin_port = htons(PORT);
    
    bind(listen_fd, (struct sockaddr *)&server_addr, sizeof(server_addr));
    listen(listen_fd, 128);
    
    // 注册监听socket的读事件
    aeCreateFileEvent(el, listen_fd, AE_READABLE, acceptTcpHandler, NULL);
    
    printf("Redis模拟服务器启动，端口 %d\n", PORT);
    
    // 启动事件循环
    aeMain(el);
    
    // 清理
    aeDeleteEventLoop(el);
    close(listen_fd);
    return 0;
}

编译运行：

# 需要Redis源码中的ae.c和ae.h
gcc -o simple_redis_ae simple_redis_ae.c ae.c -I./src -lpthread
./simple_redis_ae
# 另一个终端测试
redis-cli -p 6379 PING
# 输出: +PONG

九、效果数据：epoll vs select 压测对比

我们用 redis-benchmark 测试不同事件后端。

# 编译Redis时指定事件后端
# 默认epoll
make
# 强制使用select
make CFLAGS="-DHAVE_SELECT=1"

# 压测命令
redis-benchmark -h 127.0.0.1 -p 6379 -n 100000 -c 100 -t SET,GET

结果：

epoll在100并发下QPS高39%，P99延迟低45%。连接数越多差距越大，1000连接时select几乎不可用。

十、避坑指南

踩过的坑，写出来你们别重蹈覆辙。

10.1 坑1：epoll惊群效应

Redis 6.0之前是单线程，没有这个问题。但Redis 6.0引入多线程IO后，多个线程同时调用 epoll_wait 会导致惊群。解决方案：使用 EPOLLEXCLUSIVE 标志（Linux 4.5+）。

// 在aeApiAddEvent中设置
ee.events |= EPOLLEXCLUSIVE; // 避免惊群

10.2 坑2：文件描述符耗尽

默认 setsize=1024，如果连接数超过1024，Redis会报错 "max number of clients reached"。解决方案：修改 redis.conf 的 maxclients，同时增大系统 ulimit -n。

# 查看当前限制
ulimit -n
# 临时修改
ulimit -n 65535
# 永久修改 /etc/security/limits.conf
* soft nofile 65535
* hard nofile 65535

10.3 坑3：定时器精度问题

Redis定时器基于 gettimeofday，精度毫秒级。如果系统时间被调整（如NTP同步），可能导致定时器提前或延迟触发。解决方案：使用 clock_gettime(CLOCK_MONOTONIC) 避免系统时间影响。

// Redis 7.2.4已修复，使用monotonic时钟
// ae.c 第200行
static long long ustime(void) {
    struct timespec ts;
    clock_gettime(CLOCK_MONOTONIC, &ts);
    return ((long long)ts.tv_sec * 1000000) + (ts.tv_nsec / 1000);
}

10.4 坑4：epoll边缘触发 vs 水平触发

Redis默认使用水平触发（LT），因为代码简单。如果改成边缘触发（ET），必须用非阻塞IO并循环读取直到EAGAIN，否则会漏事件。我们试过改ET，QPS反而下降5%，因为多了一次系统调用。

// 边缘触发示例（不推荐）
ee.events |= EPOLLET; // 边缘触发
// 必须配合非阻塞IO
fcntl(fd, F_SETFL, fcntl(fd, F_GETFL) | O_NONBLOCK);

10.5 坑5：事件循环死锁

如果 beforesleep 回调中调用了 aeStop，但事件循环还在处理事件，会导致死锁。解决方案：在 beforesleep 中设置标志，循环结束后检查。

// 正确做法
void beforeSleep(aeEventLoop *el) {
    if (should_stop) {
        el->stop = 1; // 设置停止标志，循环会自然退出
    }
}

十一、总结

Redis事件驱动模型的核心是 aeEventLoop + 多路复用API。epoll在连接数多时性能碾压select/poll，但要注意惊群、fd限制、定时器精度等坑。源码不到2000行，值得每个后端工程师精读。