1. 项目概述从零到一理解Redis的核心骨架如果你对Redis的源码实现感兴趣或者想通过一个具体的项目来深入理解网络编程和事件驱动模型那么“用C/C构建自己的Redis”这个系列绝对是一个绝佳的实践路径。今天我们来到第三章实现一个回声服务器并为其注入灵魂——事件循环。这个项目的核心目标远不止是让服务器把客户端发来的消息原样返回那么简单。它的真正价值在于我们将在C/C的层面亲手搭建一个**事件循环Event Loop**的雏形。事件循环是Redis、Nginx、Node.js等高性能服务器之所以能高效处理成千上万并发连接的核心机制。通过实现一个带事件循环的回声服务器你不仅能巩固Socket编程基础更能窥见高性能服务器设计的门道理解“异步”、“非阻塞”、“IO多路复用”这些听起来高大上的概念到底是如何在代码中落地的。简单来说这个项目适合两类朋友一是希望超越基础CRUD理解系统底层原理的C/C开发者二是对Redis、Nginx等中间件内部工作机制充满好奇想知其然更知其所以然的后端工程师。我们将从最朴素的阻塞式回声服务器开始逐步揭示其性能瓶颈然后引入事件循环进行重构最终得到一个可以同时服务多个客户端、资源消耗极低的迷你版“Redis核心”。整个过程我会把每一步的“为什么”讲清楚并分享我在实现过程中踩过的坑和调试技巧。2. 核心思路拆解为什么需要事件循环在动手写代码之前我们必须想明白一个最简单的回声服务器用accept、recv、send、close这几个Socket API就能搞定为什么还要大费周章地引入事件循环答案就藏在“并发”与“效率”这两个词里。2.1 朴素阻塞服务器的性能瓶颈让我们先回顾一下一个典型的阻塞式回声服务器的流程创建Socket绑定地址开始监听。调用accept()等待客户端连接。此时主线程会阻塞直到有连接到来。连接建立后在一个循环里调用recv()读取客户端数据。recv()也会阻塞直到数据到达或连接关闭。收到数据后调用send()原样发回。然后回到第3步等待下一条消息。这种模式在单客户端测试时工作得很好。但它的致命缺陷在于“阻塞”。当服务器在accept()或recv()上等待时它什么都做不了CPU被白白闲置。如果同时有第二个客户端尝试连接它必须等第一个客户端的整个会话可能包含多次收发完全结束后服务器才能从accept()中返回处理它。这根本无法满足并发需求。一种改进是使用多线程或多进程每来一个新连接就创建一个新的线程/进程去处理它。这样主线程可以快速回到accept()继续接收新连接。这确实解决了并发问题但引入了新的开销线程/进程的创建、销毁、上下文切换成本很高。当连接数达到成千上万时这正是Redis等服务的常态系统资源会被迅速耗尽。这就是著名的 C10K问题 的由来。2.2 事件循环用一份力同时盯住多个“状态”事件循环的核心思想是非阻塞和IO多路复用。我们不让线程傻等在某一个Socket的IO操作上而是告诉操作系统“我这里有一堆Socket监听Socket和所有已连接的客户端Socket你帮我盯着点当其中任何一个变得‘可读’有数据到来或新连接或‘可写’内核发送缓冲区有空位时就通知我。”这样一个线程即事件循环就可以同时管理成百上千个连接。这个负责“盯着”和“通知”的系统调用就是IO多路复用机制在Linux上主要有三种select、poll和epoll。我们的迷你Redis将选择性能最高、目前最主流的epoll作为实现基础。所以本章的实现路线图就清晰了先实现一个阻塞式的回声服务器作为基准和对比。分析其瓶颈引入非阻塞Socket和epollAPI。设计并实现一个简单但完整的事件循环结构。将回声服务器的逻辑嵌入到事件循环中完成蜕变。3. 环境准备与基础实现工欲善其事必先利其器。我们先确保开发环境就绪并搭建一个最简单的版本作为起点。3.1 开发环境与工具链我强烈推荐使用Linux或macOS进行开发因为它们对POSIX Socket和epollmacOS对应kqueue的支持是原生的。如果你使用Windows虽然可以通过WSL2获得近乎原生的体验但为了减少环境差异带来的问题本教程的代码和命令均以Linux为准。编译器使用GCC或Clang。确保版本不要太老。# 检查gcc版本 gcc --version代码编辑器/IDEVSCode、CLion、Vim等任选。关键在于配置好代码提示和编译调试环境。对于VSCode安装C/C扩展后配置tasks.json和launch.json可以极大提升效率。调试工具gdb命令行调试神器必须掌握基础命令break,run,next,step,print,backtrace。netcat(nc)用于快速测试服务器的命令行TCP客户端。telnet另一个简单的测试客户端。strace跟踪系统调用理解程序与内核的交互过程对于调试阻塞、死锁问题非常有用。3.2 实现阻塞式回声服务器我们先写一个最基础的版本代码结构清晰但性能堪忧。这个版本将作为我们理解问题根源的标本。// blocking_echo_server.c #include stdio.h #include stdlib.h #include string.h #include unistd.h #include arpa/inet.h #include sys/socket.h #define PORT 6379 // 使用Redis默认端口增加代入感 #define BUFFER_SIZE 1024 void handle_client(int client_sock) { char buffer[BUFFER_SIZE]; ssize_t bytes_read; while ((bytes_read recv(client_sock, buffer, BUFFER_SIZE - 1, 0)) 0) { buffer[bytes_read] \0; // 确保字符串终止 printf(Received: %s, buffer); // 回声逻辑 send(client_sock, buffer, bytes_read, 0); // 简单协议如果收到quit\n则结束会话 if (strncmp(buffer, quit, 4) 0) { printf(Client requested to quit.\n); break; } } if (bytes_read 0) { printf(Client disconnected gracefully.\n); } else if (bytes_read 0) { perror(recv failed); } close(client_sock); } int main() { int server_sock, client_sock; struct sockaddr_in server_addr, client_addr; socklen_t client_addr_len sizeof(client_addr); // 1. 创建Socket server_sock socket(AF_INET, SOCK_STREAM, 0); if (server_sock 0) { perror(Socket creation failed); exit(EXIT_FAILURE); } // 2. 绑定地址 server_addr.sin_family AF_INET; server_addr.sin_addr.s_addr INADDR_ANY; server_addr.sin_port htons(PORT); if (bind(server_sock, (struct sockaddr*)server_addr, sizeof(server_addr)) 0) { perror(Bind failed); close(server_sock); exit(EXIT_FAILURE); } // 3. 开始监听 if (listen(server_sock, 5) 0) { // backlog设置为5 perror(Listen failed); close(server_sock); exit(EXIT_FAILURE); } printf(Blocking Echo Server listening on port %d...\n, PORT); // 4. 主循环接受连接并处理 while (1) { client_sock accept(server_sock, (struct sockaddr*)client_addr, client_addr_len); if (client_sock 0) { perror(Accept failed); continue; // 接受失败继续循环 } printf(New connection from %s:%d\n, inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port)); // 处理客户端连接。注意这里会阻塞直到这个客户端会话结束 handle_client(client_sock); printf(Connection from %s:%d closed.\n, inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port)); } // 理论上不会执行到这里 close(server_sock); return 0; }编译与测试gcc -o blocking_echo_server blocking_echo_server.c ./blocking_echo_server打开另一个终端使用netcat连接测试nc localhost 6379 Hello! # 输入 Hello! # 服务器返回 quit # 输入 quit # 服务器返回并断开连接注意这个服务器一次只能服务一个客户端。你可以尝试开两个nc终端第二个终端的连接会一直挂起直到第一个终端断开。这就是我们接下来要解决的核心问题。4. 引入事件循环Epoll核心机制详解要突破阻塞模型的限制我们需要请出Linux下的高性能IO多路复用利器——epoll。它比古老的select和poll更高效尤其是在连接数巨大的场景下。4.1 Epoll的三板斧epollAPI主要包含三个系统调用epoll_create1: 创建一个epoll实例返回一个文件描述符fd。这个fd将用于后续的所有操作。epoll_ctl: 用于管理epoll实例关注的文件描述符集合。可以添加EPOLL_CTL_ADD、修改EPOLL_CTL_MOD或删除EPOLL_CTL_DEL一个fd及其关注的事件如可读EPOLLIN、可写EPOLLOUT。epoll_wait: 等待事件发生。它会阻塞可以设置超时直到被监控的fd中有一个或多个发生了我们关注的事件然后将这些事件填充到提供的数组中。为什么是Epollselect和poll每次调用都需要将整个fd集合从用户空间拷贝到内核空间事件发生后又需要遍历整个集合来找出哪些fd就绪。当fd数量很多时开销线性增长。而epoll通过epoll_ctl建立fd与内核数据结构的关联epoll_wait只返回就绪的fd列表避免了不必要的拷贝和遍历时间复杂度是O(1)的。4.2 设计我们的事件循环结构一个典型的事件循环其骨架代码如下所示。我们将其封装成一个结构体便于管理状态。// event_loop.h #ifndef EVENT_LOOP_H #define EVENT_LOOP_H #include sys/epoll.h #define MAX_EVENTS 64 #define BUFFER_SIZE 1024 typedef struct { int epoll_fd; // epoll实例的文件描述符 struct epoll_event *events; // 用于接收事件的数组 int running; // 事件循环是否运行的标志 // 未来可以扩展定时器、待处理任务队列等 } EventLoop; // 初始化事件循环 EventLoop* event_loop_create(); // 运行事件循环 void event_loop_run(EventLoop *loop); // 停止事件循环 void event_loop_stop(EventLoop *loop); // 销毁事件循环释放资源 void event_loop_destroy(EventLoop *loop); // 向epoll实例添加一个文件描述符及其关注的事件 void event_loop_add_fd(EventLoop *loop, int fd, uint32_t events); // 修改epoll实例中某个文件描述符关注的事件 void event_loop_modify_fd(EventLoop *loop, int fd, uint32_t events); // 从epoll实例中删除一个文件描述符 void event_loop_remove_fd(EventLoop *loop, int fd); #endif // EVENT_LOOP_H对应的实现文件event_loop.c的核心是event_loop_run函数它包含了一个经典的epoll_wait循环// event_loop.c (部分核心代码) #include event_loop.h #include stdio.h #include stdlib.h #include unistd.h #include errno.h EventLoop* event_loop_create() { EventLoop *loop (EventLoop*)malloc(sizeof(EventLoop)); if (!loop) return NULL; loop-epoll_fd epoll_create1(0); if (loop-epoll_fd 0) { perror(epoll_create1 failed); free(loop); return NULL; } loop-events (struct epoll_event*)malloc(sizeof(struct epoll_event) * MAX_EVENTS); if (!loop-events) { perror(malloc for events failed); close(loop-epoll_fd); free(loop); return NULL; } loop-running 0; return loop; } void event_loop_run(EventLoop *loop) { if (!loop) return; loop-running 1; printf(Event loop started.\n); while (loop-running) { // 等待事件发生超时时间设为-1表示永久阻塞 int nfds epoll_wait(loop-epoll_fd, loop-events, MAX_EVENTS, -1); if (nfds 0) { // 被信号中断是正常情况继续循环 if (errno EINTR) { continue; } perror(epoll_wait failed); break; } // 处理所有就绪的事件 for (int i 0; i nfds; i) { int fd loop-events[i].data.fd; uint32_t revents loop-events[i].events; // 处理错误和挂起事件 if (revents (EPOLLERR | EPOLLHUP)) { printf(Error or hangup on fd %d\n, fd); event_loop_remove_fd(loop, fd); close(fd); continue; } // 这里只是一个框架具体的事件处理逻辑如可读、可写需要外部传入回调函数 // 我们将在回声服务器中实现具体的处理逻辑 printf(Event on fd %d, events: %u\n, fd, revents); // TODO: 根据revents调用相应的处理函数 } } printf(Event loop stopped.\n); } // ... 其他函数add_fd, remove_fd等的实现实操心得在event_loop_run的while循环中一定要检查epoll_wait的返回值。当nfds 0且errno EINTR时表示系统调用被信号中断这是正常现象例如调试器暂停应该继续循环而不是退出。忽略这一点会导致服务器在特定情况下意外终止。5. 构建非阻塞式回声服务器现在我们将事件循环与Socket结合起来。关键一步是将Socket设置为非阻塞模式这样accept、recv、send等操作在无法立即完成时会立即返回错误EAGAIN或EWOULDBLOCK而不是阻塞线程。5.1 设置Socket为非阻塞// utils.h / utils.c #include fcntl.h int set_nonblocking(int fd) { int flags fcntl(fd, F_GETFL, 0); if (flags -1) return -1; return fcntl(fd, F_SETFL, flags | O_NONBLOCK); }5.2 整合事件循环与Socket处理我们创建一个新的服务器文件echo_server_with_epoll.c。它的核心逻辑是创建监听Socket设置为非阻塞并添加到epoll实例关注EPOLLIN可读事件。在事件循环中当监听Socket可读时调用accept接受新连接。将新连接的客户端Socket也设置为非阻塞并添加到epoll实例关注EPOLLIN事件。当客户端Socket可读时调用recv读取数据。这里需要循环读取因为非阻塞模式下一次recv可能读不完所有数据。将读到的数据放入该客户端的缓冲区。为了简化我们假设一次recv就能读完一条消息然后立即关注EPOLLOUT可写事件准备回送。当客户端Socket可写时调用send发送缓冲区中的数据。同样需要循环发送直到数据全部写完或遇到EAGAIN。发送完毕后重新关注EPOLLIN事件等待下一条消息。如果客户端断开连接recv返回0则从epoll中移除并关闭Socket。以下是核心部分的代码框架// echo_server_with_epoll.c (核心部分) #include event_loop.h #include utils.h // 包含set_nonblocking等 #include string.h #include arpa/inet.h // 为每个客户端连接定义一个上下文结构 typedef struct { int fd; char buffer[BUFFER_SIZE]; size_t buf_len; // 缓冲区中数据的长度 size_t sent_len; // 已发送数据的长度用于处理发送不全的情况 int want_write; // 是否需要关注可写事件 } ClientContext; // 全局的客户端上下文表简单起见用数组生产环境应用哈希表等 ClientContext clients[FD_SETSIZE]; // FD_SETSIZE是系统定义的fd最大数量常数 void handle_accept_event(EventLoop *loop, int server_fd) { struct sockaddr_in client_addr; socklen_t addr_len sizeof(client_addr); int client_fd; // 非阻塞accept可能需要循环接受多个连接 while ((client_fd accept(server_fd, (struct sockaddr*)client_addr, addr_len)) 0) { printf(Accepted new connection, fd%d, from %s:%d\n, client_fd, inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port)); if (set_nonblocking(client_fd) 0) { perror(set_nonblocking failed for client); close(client_fd); continue; } // 初始化客户端上下文 clients[client_fd].fd client_fd; clients[client_fd].buf_len 0; clients[client_fd].sent_len 0; clients[client_fd].want_write 0; // 添加到epoll初始只关注可读事件 event_loop_add_fd(loop, client_fd, EPOLLIN | EPOLLET); // EPOLLET 表示边缘触发模式后面会讲 } if (client_fd 0 errno ! EAGAIN errno ! EWOULDBLOCK) { perror(accept failed); } } void handle_read_event(EventLoop *loop, int client_fd) { ClientContext *ctx clients[client_fd]; ssize_t n recv(client_fd, ctx-buffer, BUFFER_SIZE - 1, 0); if (n 0) { ctx-buffer[n] \0; ctx-buf_len n; ctx-sent_len 0; printf(Received from fd%d: %s\n, client_fd, ctx-buffer); // 检查是否是退出命令 if (strncmp(ctx-buffer, quit, 4) 0) { printf(Client fd%d requested to quit.\n, client_fd); event_loop_remove_fd(loop, client_fd); close(client_fd); memset(ctx, 0, sizeof(ClientContext)); // 清理上下文 return; } // 现在有数据要发送需要关注可写事件 ctx-want_write 1; // 修改epoll关注的事件增加EPOLLOUT event_loop_modify_fd(loop, client_fd, EPOLLIN | EPOLLOUT | EPOLLET); } else if (n 0) { // 客户端正常关闭连接 printf(Client fd%d disconnected.\n, client_fd); event_loop_remove_fd(loop, client_fd); close(client_fd); memset(ctx, 0, sizeof(ClientContext)); } else { // 错误处理 if (errno EAGAIN || errno EWOULDBLOCK) { // 非阻塞模式下数据还没准备好下次再读 return; } perror(recv error); event_loop_remove_fd(loop, client_fd); close(client_fd); memset(ctx, 0, sizeof(ClientContext)); } } void handle_write_event(EventLoop *loop, int client_fd) { ClientContext *ctx clients[client_fd]; if (!ctx-want_write) { // 没有数据要写可能是事件误触发移除可写关注 event_loop_modify_fd(loop, client_fd, EPOLLIN | EPOLLET); return; } ssize_t n send(client_fd, ctx-buffer ctx-sent_len, ctx-buf_len - ctx-sent_len, 0); if (n 0) { ctx-sent_len n; if (ctx-sent_len ctx-buf_len) { // 数据全部发送完毕 printf(Echoed back to fd%d.\n, client_fd); ctx-want_write 0; ctx-buf_len 0; ctx-sent_len 0; // 发送完毕只关注可读事件准备接收下一条消息 event_loop_modify_fd(loop, client_fd, EPOLLIN | EPOLLET); } // 如果没发完下次可写事件触发时会继续发送 } else if (n 0) { if (errno EAGAIN || errno EWOULDBLOCK) { // 内核发送缓冲区已满下次再试 return; } perror(send error); event_loop_remove_fd(loop, client_fd); close(client_fd); memset(ctx, 0, sizeof(ClientContext)); } // n0 在send中一般不会发生 } // 修改event_loop_run中的事件分发逻辑 void event_loop_run_with_handler(EventLoop *loop, int server_fd) { // ... 初始化 ... while (loop-running) { int nfds epoll_wait(loop-epoll_fd, loop-events, MAX_EVENTS, -1); // ... 错误处理 ... for (int i 0; i nfds; i) { int fd loop-events[i].data.fd; uint32_t revents loop-events[i].events; if (revents (EPOLLERR | EPOLLHUP)) { // ... 错误处理关闭连接 ... continue; } if (fd server_fd) { // 监听socket有可读事件表示有新连接 handle_accept_event(loop, server_fd); } else { // 客户端socket事件 if (revents EPOLLIN) { handle_read_event(loop, fd); } if (revents EPOLLOUT) { handle_write_event(loop, fd); } } } } }主函数负责创建监听Socket、初始化事件循环并启动它int main() { // 创建、绑定、监听Socket与阻塞版本类似但创建后要设置为非阻塞 int server_fd socket(AF_INET, SOCK_STREAM, 0); // ... 绑定和监听 ... set_nonblocking(server_fd); // 关键 EventLoop *loop event_loop_create(); if (!loop) { fprintf(stderr, Failed to create event loop.\n); exit(EXIT_FAILURE); } // 将监听Socket加入事件循环关注可读事件 event_loop_add_fd(loop, server_fd, EPOLLIN); printf(Non-blocking Echo Server with Event Loop listening on port %d...\n, PORT); // 运行带事件处理器的事件循环 event_loop_run_with_handler(loop, server_fd); // 清理 event_loop_destroy(loop); close(server_fd); return 0; }5.3 边缘触发(ET)与水平触发(LT)模式在上面的代码中你注意到了EPOLLET这个标志吗这代表使用**边缘触发Edge-Triggered**模式。epoll有两种工作模式水平触发LT默认只要文件描述符处于就绪状态例如接收缓冲区有数据每次调用epoll_wait都会报告该事件。边缘触发ET只有当文件描述符状态发生变化时例如从无数据到有数据才会报告一次事件。选择ET还是LTLT模式编程更简单不容易遗漏事件。如果你收到一个可读事件后没有一次性读完所有数据下次epoll_wait还会通知你。但这也可能导致效率问题如果数据一直没读完就会一直被通知。ET模式效率更高因为它减少了事件通知的次数。但编程复杂度陡增你必须一次性把数据读完或写到EAGAIN为止否则可能会永远丢失这次事件。上面的示例代码使用了ET模式因此在handle_read_event和handle_write_event中我们都需要用循环来确保处理完所有可用数据或空间。重要提示使用ET模式时必须将文件描述符设置为非阻塞。否则在循环读/写到最后时可能会在最后一次阻塞住导致事件循环卡死。这是初学者最容易踩的坑。6. 测试、调试与性能对比6.1 功能测试编译并运行我们的非阻塞服务器gcc -o epoll_echo_server event_loop.c echo_server_with_epoll.c utils.c ./epoll_echo_server打开多个终端用nc或telnet连接测试。你会发现现在服务器可以同时与多个客户端进行交互了每个客户端的输入都能独立地得到回声响应。6.2 压力测试与性能观测我们可以用简单的脚本或工具进行并发测试。例如使用ab(Apache Bench)进行HTTP压力测试需要稍改服务器以支持HTTP响应头或者用Python写一个简单的多客户端测试脚本。更直观的方法是观察资源占用。在服务器运行期间打开另一个终端使用top或htop命令查看CPU和内存占用。你会发现即使有几十个空闲连接这个单进程、单线程的服务器CPU占用率也几乎为0内存增长也非常有限。这与为每个连接创建一个线程的模型形成鲜明对比。6.3 常见问题与调试技巧在实现过程中你几乎一定会遇到下面这些问题accept: Invalid argument或Bad file descriptor原因通常是在客户端Socket已经关闭后仍然尝试对其进行操作或者epoll_ctl操作了无效的fd。排查在每次close(fd)后立即将对应的上下文清空并在操作fd前检查其有效性。使用strace跟踪系统调用可以清晰看到fd的生命周期。服务器CPU占用率100%原因事件循环空转。最常见于ET模式下某个Socket一直处于可读或可写状态导致epoll_wait不断立即返回。排查检查你的读/写处理逻辑是否做到了“耗尽”状态。对于可读事件必须循环recv直到返回EAGAIN。对于可写事件通常只在有数据要发送时才关注EPOLLOUT发送完毕后应立即取消关注。客户端收不到完整数据或数据粘连原因TCP是字节流协议没有消息边界。recv一次可能收到多条客户端消息也可能一条消息分多次收到。解决这是网络编程的经典问题。需要在应用层定义协议。对于我们的回声服务器一个简单的方法是以换行符\n作为消息分隔符。在handle_read_event中需要解析缓冲区按\n切分消息。Redis自己的协议RESP就是一套完整的应用层协议。内存泄漏原因malloc的ClientContext或事件数组没有free或者客户端断开后上下文没有清理干净。排查使用valgrind工具进行内存检查valgrind --leak-checkfull ./epoll_echo_server。使用GDB调试事件驱动程序事件循环程序因为一直在循环打断点不太方便。可以在epoll_wait处打断点当有事件时程序会停在这里。使用info threads查看线程虽然我们是单线程使用bt查看调用栈。使用p variable打印变量特别是errno的值在系统调用失败后立即打印errno能快速定位问题。7. 从回声服务器到迷你Redis的思考至此我们已经成功构建了一个基于事件循环的高性能并发回声服务器。但这距离一个真正的Redis还非常遥远。我们可以沿着这个架构思考下一步的演进方向协议解析器实现Redis序列化协议RESP的解析。这需要替换掉简单的按行读取编写一个状态机来解析*3\r\n$3\r\nSET\r\n$5\r\nmykey\r\n$7\r\nmyvalue\r\n这样的格式。命令处理根据解析出的命令如SET、GET、DEL操作内存中的数据结构。这就需要实现一个高效的键值存储可能涉及哈希表、跳表等。数据持久化实现RDB快照和AOF日志这是Redis可靠性的关键。多线程优化Redis 6.0引入了多线程IO但事件循环主线程负责命令执行。我们可以思考如何将网络IOread/write卸载到工作线程而主线程只负责epoll_wait和命令解析。更完善的事件循环添加定时器事件用于处理键过期、持久化调度等、任务队列用于处理耗时的阻塞操作如持久化文件IO避免阻塞事件循环。实现一个完整Redis的工程量是巨大的但通过这个回声服务器项目你已经抓住了它最核心、最精华的部分——单线程事件循环模型。理解了这一点你再去看Redis的源码ae.c,ae_epoll.c等文件就会有一种豁然开朗的感觉。你不再是被动地接受“Redis是单线程的”这个结论而是从底层明白了它为什么可以这么设计以及这种设计带来的巨大优势和需要规避的陷阱。我个人在实现这个项目的过程中最大的收获不是写出了多少行代码而是对“非阻塞”、“异步”、“事件驱动”这些概念建立了肌肉记忆。当你亲手处理了EAGAIN错误当你调试了ET模式下的数据读取循环当你看到单线程轻松管理上百个连接时这些知识就真正属于你了。下次当你遇到Nginx配置、Node.js回调或者任何高并发场景时你脑海中的第一反应将是“哦这背后大概是一个事件循环在调度”。这种透过现象看本质的能力才是这个项目带给你的最大价值。
1. 项目概述从零到一理解Redis的核心骨架如果你对Redis的源码实现感兴趣或者想通过一个具体的项目来深入理解网络编程和事件驱动模型那么“用C/C构建自己的Redis”这个系列绝对是一个绝佳的实践路径。今天我们来到第三章实现一个回声服务器并为其注入灵魂——事件循环。这个项目的核心目标远不止是让服务器把客户端发来的消息原样返回那么简单。它的真正价值在于我们将在C/C的层面亲手搭建一个**事件循环Event Loop**的雏形。事件循环是Redis、Nginx、Node.js等高性能服务器之所以能高效处理成千上万并发连接的核心机制。通过实现一个带事件循环的回声服务器你不仅能巩固Socket编程基础更能窥见高性能服务器设计的门道理解“异步”、“非阻塞”、“IO多路复用”这些听起来高大上的概念到底是如何在代码中落地的。简单来说这个项目适合两类朋友一是希望超越基础CRUD理解系统底层原理的C/C开发者二是对Redis、Nginx等中间件内部工作机制充满好奇想知其然更知其所以然的后端工程师。我们将从最朴素的阻塞式回声服务器开始逐步揭示其性能瓶颈然后引入事件循环进行重构最终得到一个可以同时服务多个客户端、资源消耗极低的迷你版“Redis核心”。整个过程我会把每一步的“为什么”讲清楚并分享我在实现过程中踩过的坑和调试技巧。2. 核心思路拆解为什么需要事件循环在动手写代码之前我们必须想明白一个最简单的回声服务器用accept、recv、send、close这几个Socket API就能搞定为什么还要大费周章地引入事件循环答案就藏在“并发”与“效率”这两个词里。2.1 朴素阻塞服务器的性能瓶颈让我们先回顾一下一个典型的阻塞式回声服务器的流程创建Socket绑定地址开始监听。调用accept()等待客户端连接。此时主线程会阻塞直到有连接到来。连接建立后在一个循环里调用recv()读取客户端数据。recv()也会阻塞直到数据到达或连接关闭。收到数据后调用send()原样发回。然后回到第3步等待下一条消息。这种模式在单客户端测试时工作得很好。但它的致命缺陷在于“阻塞”。当服务器在accept()或recv()上等待时它什么都做不了CPU被白白闲置。如果同时有第二个客户端尝试连接它必须等第一个客户端的整个会话可能包含多次收发完全结束后服务器才能从accept()中返回处理它。这根本无法满足并发需求。一种改进是使用多线程或多进程每来一个新连接就创建一个新的线程/进程去处理它。这样主线程可以快速回到accept()继续接收新连接。这确实解决了并发问题但引入了新的开销线程/进程的创建、销毁、上下文切换成本很高。当连接数达到成千上万时这正是Redis等服务的常态系统资源会被迅速耗尽。这就是著名的 C10K问题 的由来。2.2 事件循环用一份力同时盯住多个“状态”事件循环的核心思想是非阻塞和IO多路复用。我们不让线程傻等在某一个Socket的IO操作上而是告诉操作系统“我这里有一堆Socket监听Socket和所有已连接的客户端Socket你帮我盯着点当其中任何一个变得‘可读’有数据到来或新连接或‘可写’内核发送缓冲区有空位时就通知我。”这样一个线程即事件循环就可以同时管理成百上千个连接。这个负责“盯着”和“通知”的系统调用就是IO多路复用机制在Linux上主要有三种select、poll和epoll。我们的迷你Redis将选择性能最高、目前最主流的epoll作为实现基础。所以本章的实现路线图就清晰了先实现一个阻塞式的回声服务器作为基准和对比。分析其瓶颈引入非阻塞Socket和epollAPI。设计并实现一个简单但完整的事件循环结构。将回声服务器的逻辑嵌入到事件循环中完成蜕变。3. 环境准备与基础实现工欲善其事必先利其器。我们先确保开发环境就绪并搭建一个最简单的版本作为起点。3.1 开发环境与工具链我强烈推荐使用Linux或macOS进行开发因为它们对POSIX Socket和epollmacOS对应kqueue的支持是原生的。如果你使用Windows虽然可以通过WSL2获得近乎原生的体验但为了减少环境差异带来的问题本教程的代码和命令均以Linux为准。编译器使用GCC或Clang。确保版本不要太老。# 检查gcc版本 gcc --version代码编辑器/IDEVSCode、CLion、Vim等任选。关键在于配置好代码提示和编译调试环境。对于VSCode安装C/C扩展后配置tasks.json和launch.json可以极大提升效率。调试工具gdb命令行调试神器必须掌握基础命令break,run,next,step,print,backtrace。netcat(nc)用于快速测试服务器的命令行TCP客户端。telnet另一个简单的测试客户端。strace跟踪系统调用理解程序与内核的交互过程对于调试阻塞、死锁问题非常有用。3.2 实现阻塞式回声服务器我们先写一个最基础的版本代码结构清晰但性能堪忧。这个版本将作为我们理解问题根源的标本。// blocking_echo_server.c #include stdio.h #include stdlib.h #include string.h #include unistd.h #include arpa/inet.h #include sys/socket.h #define PORT 6379 // 使用Redis默认端口增加代入感 #define BUFFER_SIZE 1024 void handle_client(int client_sock) { char buffer[BUFFER_SIZE]; ssize_t bytes_read; while ((bytes_read recv(client_sock, buffer, BUFFER_SIZE - 1, 0)) 0) { buffer[bytes_read] \0; // 确保字符串终止 printf(Received: %s, buffer); // 回声逻辑 send(client_sock, buffer, bytes_read, 0); // 简单协议如果收到quit\n则结束会话 if (strncmp(buffer, quit, 4) 0) { printf(Client requested to quit.\n); break; } } if (bytes_read 0) { printf(Client disconnected gracefully.\n); } else if (bytes_read 0) { perror(recv failed); } close(client_sock); } int main() { int server_sock, client_sock; struct sockaddr_in server_addr, client_addr; socklen_t client_addr_len sizeof(client_addr); // 1. 创建Socket server_sock socket(AF_INET, SOCK_STREAM, 0); if (server_sock 0) { perror(Socket creation failed); exit(EXIT_FAILURE); } // 2. 绑定地址 server_addr.sin_family AF_INET; server_addr.sin_addr.s_addr INADDR_ANY; server_addr.sin_port htons(PORT); if (bind(server_sock, (struct sockaddr*)server_addr, sizeof(server_addr)) 0) { perror(Bind failed); close(server_sock); exit(EXIT_FAILURE); } // 3. 开始监听 if (listen(server_sock, 5) 0) { // backlog设置为5 perror(Listen failed); close(server_sock); exit(EXIT_FAILURE); } printf(Blocking Echo Server listening on port %d...\n, PORT); // 4. 主循环接受连接并处理 while (1) { client_sock accept(server_sock, (struct sockaddr*)client_addr, client_addr_len); if (client_sock 0) { perror(Accept failed); continue; // 接受失败继续循环 } printf(New connection from %s:%d\n, inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port)); // 处理客户端连接。注意这里会阻塞直到这个客户端会话结束 handle_client(client_sock); printf(Connection from %s:%d closed.\n, inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port)); } // 理论上不会执行到这里 close(server_sock); return 0; }编译与测试gcc -o blocking_echo_server blocking_echo_server.c ./blocking_echo_server打开另一个终端使用netcat连接测试nc localhost 6379 Hello! # 输入 Hello! # 服务器返回 quit # 输入 quit # 服务器返回并断开连接注意这个服务器一次只能服务一个客户端。你可以尝试开两个nc终端第二个终端的连接会一直挂起直到第一个终端断开。这就是我们接下来要解决的核心问题。4. 引入事件循环Epoll核心机制详解要突破阻塞模型的限制我们需要请出Linux下的高性能IO多路复用利器——epoll。它比古老的select和poll更高效尤其是在连接数巨大的场景下。4.1 Epoll的三板斧epollAPI主要包含三个系统调用epoll_create1: 创建一个epoll实例返回一个文件描述符fd。这个fd将用于后续的所有操作。epoll_ctl: 用于管理epoll实例关注的文件描述符集合。可以添加EPOLL_CTL_ADD、修改EPOLL_CTL_MOD或删除EPOLL_CTL_DEL一个fd及其关注的事件如可读EPOLLIN、可写EPOLLOUT。epoll_wait: 等待事件发生。它会阻塞可以设置超时直到被监控的fd中有一个或多个发生了我们关注的事件然后将这些事件填充到提供的数组中。为什么是Epollselect和poll每次调用都需要将整个fd集合从用户空间拷贝到内核空间事件发生后又需要遍历整个集合来找出哪些fd就绪。当fd数量很多时开销线性增长。而epoll通过epoll_ctl建立fd与内核数据结构的关联epoll_wait只返回就绪的fd列表避免了不必要的拷贝和遍历时间复杂度是O(1)的。4.2 设计我们的事件循环结构一个典型的事件循环其骨架代码如下所示。我们将其封装成一个结构体便于管理状态。// event_loop.h #ifndef EVENT_LOOP_H #define EVENT_LOOP_H #include sys/epoll.h #define MAX_EVENTS 64 #define BUFFER_SIZE 1024 typedef struct { int epoll_fd; // epoll实例的文件描述符 struct epoll_event *events; // 用于接收事件的数组 int running; // 事件循环是否运行的标志 // 未来可以扩展定时器、待处理任务队列等 } EventLoop; // 初始化事件循环 EventLoop* event_loop_create(); // 运行事件循环 void event_loop_run(EventLoop *loop); // 停止事件循环 void event_loop_stop(EventLoop *loop); // 销毁事件循环释放资源 void event_loop_destroy(EventLoop *loop); // 向epoll实例添加一个文件描述符及其关注的事件 void event_loop_add_fd(EventLoop *loop, int fd, uint32_t events); // 修改epoll实例中某个文件描述符关注的事件 void event_loop_modify_fd(EventLoop *loop, int fd, uint32_t events); // 从epoll实例中删除一个文件描述符 void event_loop_remove_fd(EventLoop *loop, int fd); #endif // EVENT_LOOP_H对应的实现文件event_loop.c的核心是event_loop_run函数它包含了一个经典的epoll_wait循环// event_loop.c (部分核心代码) #include event_loop.h #include stdio.h #include stdlib.h #include unistd.h #include errno.h EventLoop* event_loop_create() { EventLoop *loop (EventLoop*)malloc(sizeof(EventLoop)); if (!loop) return NULL; loop-epoll_fd epoll_create1(0); if (loop-epoll_fd 0) { perror(epoll_create1 failed); free(loop); return NULL; } loop-events (struct epoll_event*)malloc(sizeof(struct epoll_event) * MAX_EVENTS); if (!loop-events) { perror(malloc for events failed); close(loop-epoll_fd); free(loop); return NULL; } loop-running 0; return loop; } void event_loop_run(EventLoop *loop) { if (!loop) return; loop-running 1; printf(Event loop started.\n); while (loop-running) { // 等待事件发生超时时间设为-1表示永久阻塞 int nfds epoll_wait(loop-epoll_fd, loop-events, MAX_EVENTS, -1); if (nfds 0) { // 被信号中断是正常情况继续循环 if (errno EINTR) { continue; } perror(epoll_wait failed); break; } // 处理所有就绪的事件 for (int i 0; i nfds; i) { int fd loop-events[i].data.fd; uint32_t revents loop-events[i].events; // 处理错误和挂起事件 if (revents (EPOLLERR | EPOLLHUP)) { printf(Error or hangup on fd %d\n, fd); event_loop_remove_fd(loop, fd); close(fd); continue; } // 这里只是一个框架具体的事件处理逻辑如可读、可写需要外部传入回调函数 // 我们将在回声服务器中实现具体的处理逻辑 printf(Event on fd %d, events: %u\n, fd, revents); // TODO: 根据revents调用相应的处理函数 } } printf(Event loop stopped.\n); } // ... 其他函数add_fd, remove_fd等的实现实操心得在event_loop_run的while循环中一定要检查epoll_wait的返回值。当nfds 0且errno EINTR时表示系统调用被信号中断这是正常现象例如调试器暂停应该继续循环而不是退出。忽略这一点会导致服务器在特定情况下意外终止。5. 构建非阻塞式回声服务器现在我们将事件循环与Socket结合起来。关键一步是将Socket设置为非阻塞模式这样accept、recv、send等操作在无法立即完成时会立即返回错误EAGAIN或EWOULDBLOCK而不是阻塞线程。5.1 设置Socket为非阻塞// utils.h / utils.c #include fcntl.h int set_nonblocking(int fd) { int flags fcntl(fd, F_GETFL, 0); if (flags -1) return -1; return fcntl(fd, F_SETFL, flags | O_NONBLOCK); }5.2 整合事件循环与Socket处理我们创建一个新的服务器文件echo_server_with_epoll.c。它的核心逻辑是创建监听Socket设置为非阻塞并添加到epoll实例关注EPOLLIN可读事件。在事件循环中当监听Socket可读时调用accept接受新连接。将新连接的客户端Socket也设置为非阻塞并添加到epoll实例关注EPOLLIN事件。当客户端Socket可读时调用recv读取数据。这里需要循环读取因为非阻塞模式下一次recv可能读不完所有数据。将读到的数据放入该客户端的缓冲区。为了简化我们假设一次recv就能读完一条消息然后立即关注EPOLLOUT可写事件准备回送。当客户端Socket可写时调用send发送缓冲区中的数据。同样需要循环发送直到数据全部写完或遇到EAGAIN。发送完毕后重新关注EPOLLIN事件等待下一条消息。如果客户端断开连接recv返回0则从epoll中移除并关闭Socket。以下是核心部分的代码框架// echo_server_with_epoll.c (核心部分) #include event_loop.h #include utils.h // 包含set_nonblocking等 #include string.h #include arpa/inet.h // 为每个客户端连接定义一个上下文结构 typedef struct { int fd; char buffer[BUFFER_SIZE]; size_t buf_len; // 缓冲区中数据的长度 size_t sent_len; // 已发送数据的长度用于处理发送不全的情况 int want_write; // 是否需要关注可写事件 } ClientContext; // 全局的客户端上下文表简单起见用数组生产环境应用哈希表等 ClientContext clients[FD_SETSIZE]; // FD_SETSIZE是系统定义的fd最大数量常数 void handle_accept_event(EventLoop *loop, int server_fd) { struct sockaddr_in client_addr; socklen_t addr_len sizeof(client_addr); int client_fd; // 非阻塞accept可能需要循环接受多个连接 while ((client_fd accept(server_fd, (struct sockaddr*)client_addr, addr_len)) 0) { printf(Accepted new connection, fd%d, from %s:%d\n, client_fd, inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port)); if (set_nonblocking(client_fd) 0) { perror(set_nonblocking failed for client); close(client_fd); continue; } // 初始化客户端上下文 clients[client_fd].fd client_fd; clients[client_fd].buf_len 0; clients[client_fd].sent_len 0; clients[client_fd].want_write 0; // 添加到epoll初始只关注可读事件 event_loop_add_fd(loop, client_fd, EPOLLIN | EPOLLET); // EPOLLET 表示边缘触发模式后面会讲 } if (client_fd 0 errno ! EAGAIN errno ! EWOULDBLOCK) { perror(accept failed); } } void handle_read_event(EventLoop *loop, int client_fd) { ClientContext *ctx clients[client_fd]; ssize_t n recv(client_fd, ctx-buffer, BUFFER_SIZE - 1, 0); if (n 0) { ctx-buffer[n] \0; ctx-buf_len n; ctx-sent_len 0; printf(Received from fd%d: %s\n, client_fd, ctx-buffer); // 检查是否是退出命令 if (strncmp(ctx-buffer, quit, 4) 0) { printf(Client fd%d requested to quit.\n, client_fd); event_loop_remove_fd(loop, client_fd); close(client_fd); memset(ctx, 0, sizeof(ClientContext)); // 清理上下文 return; } // 现在有数据要发送需要关注可写事件 ctx-want_write 1; // 修改epoll关注的事件增加EPOLLOUT event_loop_modify_fd(loop, client_fd, EPOLLIN | EPOLLOUT | EPOLLET); } else if (n 0) { // 客户端正常关闭连接 printf(Client fd%d disconnected.\n, client_fd); event_loop_remove_fd(loop, client_fd); close(client_fd); memset(ctx, 0, sizeof(ClientContext)); } else { // 错误处理 if (errno EAGAIN || errno EWOULDBLOCK) { // 非阻塞模式下数据还没准备好下次再读 return; } perror(recv error); event_loop_remove_fd(loop, client_fd); close(client_fd); memset(ctx, 0, sizeof(ClientContext)); } } void handle_write_event(EventLoop *loop, int client_fd) { ClientContext *ctx clients[client_fd]; if (!ctx-want_write) { // 没有数据要写可能是事件误触发移除可写关注 event_loop_modify_fd(loop, client_fd, EPOLLIN | EPOLLET); return; } ssize_t n send(client_fd, ctx-buffer ctx-sent_len, ctx-buf_len - ctx-sent_len, 0); if (n 0) { ctx-sent_len n; if (ctx-sent_len ctx-buf_len) { // 数据全部发送完毕 printf(Echoed back to fd%d.\n, client_fd); ctx-want_write 0; ctx-buf_len 0; ctx-sent_len 0; // 发送完毕只关注可读事件准备接收下一条消息 event_loop_modify_fd(loop, client_fd, EPOLLIN | EPOLLET); } // 如果没发完下次可写事件触发时会继续发送 } else if (n 0) { if (errno EAGAIN || errno EWOULDBLOCK) { // 内核发送缓冲区已满下次再试 return; } perror(send error); event_loop_remove_fd(loop, client_fd); close(client_fd); memset(ctx, 0, sizeof(ClientContext)); } // n0 在send中一般不会发生 } // 修改event_loop_run中的事件分发逻辑 void event_loop_run_with_handler(EventLoop *loop, int server_fd) { // ... 初始化 ... while (loop-running) { int nfds epoll_wait(loop-epoll_fd, loop-events, MAX_EVENTS, -1); // ... 错误处理 ... for (int i 0; i nfds; i) { int fd loop-events[i].data.fd; uint32_t revents loop-events[i].events; if (revents (EPOLLERR | EPOLLHUP)) { // ... 错误处理关闭连接 ... continue; } if (fd server_fd) { // 监听socket有可读事件表示有新连接 handle_accept_event(loop, server_fd); } else { // 客户端socket事件 if (revents EPOLLIN) { handle_read_event(loop, fd); } if (revents EPOLLOUT) { handle_write_event(loop, fd); } } } } }主函数负责创建监听Socket、初始化事件循环并启动它int main() { // 创建、绑定、监听Socket与阻塞版本类似但创建后要设置为非阻塞 int server_fd socket(AF_INET, SOCK_STREAM, 0); // ... 绑定和监听 ... set_nonblocking(server_fd); // 关键 EventLoop *loop event_loop_create(); if (!loop) { fprintf(stderr, Failed to create event loop.\n); exit(EXIT_FAILURE); } // 将监听Socket加入事件循环关注可读事件 event_loop_add_fd(loop, server_fd, EPOLLIN); printf(Non-blocking Echo Server with Event Loop listening on port %d...\n, PORT); // 运行带事件处理器的事件循环 event_loop_run_with_handler(loop, server_fd); // 清理 event_loop_destroy(loop); close(server_fd); return 0; }5.3 边缘触发(ET)与水平触发(LT)模式在上面的代码中你注意到了EPOLLET这个标志吗这代表使用**边缘触发Edge-Triggered**模式。epoll有两种工作模式水平触发LT默认只要文件描述符处于就绪状态例如接收缓冲区有数据每次调用epoll_wait都会报告该事件。边缘触发ET只有当文件描述符状态发生变化时例如从无数据到有数据才会报告一次事件。选择ET还是LTLT模式编程更简单不容易遗漏事件。如果你收到一个可读事件后没有一次性读完所有数据下次epoll_wait还会通知你。但这也可能导致效率问题如果数据一直没读完就会一直被通知。ET模式效率更高因为它减少了事件通知的次数。但编程复杂度陡增你必须一次性把数据读完或写到EAGAIN为止否则可能会永远丢失这次事件。上面的示例代码使用了ET模式因此在handle_read_event和handle_write_event中我们都需要用循环来确保处理完所有可用数据或空间。重要提示使用ET模式时必须将文件描述符设置为非阻塞。否则在循环读/写到最后时可能会在最后一次阻塞住导致事件循环卡死。这是初学者最容易踩的坑。6. 测试、调试与性能对比6.1 功能测试编译并运行我们的非阻塞服务器gcc -o epoll_echo_server event_loop.c echo_server_with_epoll.c utils.c ./epoll_echo_server打开多个终端用nc或telnet连接测试。你会发现现在服务器可以同时与多个客户端进行交互了每个客户端的输入都能独立地得到回声响应。6.2 压力测试与性能观测我们可以用简单的脚本或工具进行并发测试。例如使用ab(Apache Bench)进行HTTP压力测试需要稍改服务器以支持HTTP响应头或者用Python写一个简单的多客户端测试脚本。更直观的方法是观察资源占用。在服务器运行期间打开另一个终端使用top或htop命令查看CPU和内存占用。你会发现即使有几十个空闲连接这个单进程、单线程的服务器CPU占用率也几乎为0内存增长也非常有限。这与为每个连接创建一个线程的模型形成鲜明对比。6.3 常见问题与调试技巧在实现过程中你几乎一定会遇到下面这些问题accept: Invalid argument或Bad file descriptor原因通常是在客户端Socket已经关闭后仍然尝试对其进行操作或者epoll_ctl操作了无效的fd。排查在每次close(fd)后立即将对应的上下文清空并在操作fd前检查其有效性。使用strace跟踪系统调用可以清晰看到fd的生命周期。服务器CPU占用率100%原因事件循环空转。最常见于ET模式下某个Socket一直处于可读或可写状态导致epoll_wait不断立即返回。排查检查你的读/写处理逻辑是否做到了“耗尽”状态。对于可读事件必须循环recv直到返回EAGAIN。对于可写事件通常只在有数据要发送时才关注EPOLLOUT发送完毕后应立即取消关注。客户端收不到完整数据或数据粘连原因TCP是字节流协议没有消息边界。recv一次可能收到多条客户端消息也可能一条消息分多次收到。解决这是网络编程的经典问题。需要在应用层定义协议。对于我们的回声服务器一个简单的方法是以换行符\n作为消息分隔符。在handle_read_event中需要解析缓冲区按\n切分消息。Redis自己的协议RESP就是一套完整的应用层协议。内存泄漏原因malloc的ClientContext或事件数组没有free或者客户端断开后上下文没有清理干净。排查使用valgrind工具进行内存检查valgrind --leak-checkfull ./epoll_echo_server。使用GDB调试事件驱动程序事件循环程序因为一直在循环打断点不太方便。可以在epoll_wait处打断点当有事件时程序会停在这里。使用info threads查看线程虽然我们是单线程使用bt查看调用栈。使用p variable打印变量特别是errno的值在系统调用失败后立即打印errno能快速定位问题。7. 从回声服务器到迷你Redis的思考至此我们已经成功构建了一个基于事件循环的高性能并发回声服务器。但这距离一个真正的Redis还非常遥远。我们可以沿着这个架构思考下一步的演进方向协议解析器实现Redis序列化协议RESP的解析。这需要替换掉简单的按行读取编写一个状态机来解析*3\r\n$3\r\nSET\r\n$5\r\nmykey\r\n$7\r\nmyvalue\r\n这样的格式。命令处理根据解析出的命令如SET、GET、DEL操作内存中的数据结构。这就需要实现一个高效的键值存储可能涉及哈希表、跳表等。数据持久化实现RDB快照和AOF日志这是Redis可靠性的关键。多线程优化Redis 6.0引入了多线程IO但事件循环主线程负责命令执行。我们可以思考如何将网络IOread/write卸载到工作线程而主线程只负责epoll_wait和命令解析。更完善的事件循环添加定时器事件用于处理键过期、持久化调度等、任务队列用于处理耗时的阻塞操作如持久化文件IO避免阻塞事件循环。实现一个完整Redis的工程量是巨大的但通过这个回声服务器项目你已经抓住了它最核心、最精华的部分——单线程事件循环模型。理解了这一点你再去看Redis的源码ae.c,ae_epoll.c等文件就会有一种豁然开朗的感觉。你不再是被动地接受“Redis是单线程的”这个结论而是从底层明白了它为什么可以这么设计以及这种设计带来的巨大优势和需要规避的陷阱。我个人在实现这个项目的过程中最大的收获不是写出了多少行代码而是对“非阻塞”、“异步”、“事件驱动”这些概念建立了肌肉记忆。当你亲手处理了EAGAIN错误当你调试了ET模式下的数据读取循环当你看到单线程轻松管理上百个连接时这些知识就真正属于你了。下次当你遇到Nginx配置、Node.js回调或者任何高并发场景时你脑海中的第一反应将是“哦这背后大概是一个事件循环在调度”。这种透过现象看本质的能力才是这个项目带给你的最大价值。
![[C++]内存管理：串顺序存储的内存回收](http://pic.xiahunao.cn/yaotu/[C++]内存管理：串顺序存储的内存回收)
