1. 为什么“事件驱动”能替代“一连接一线程”这个老思路1.1 从最简单的 accept 阻塞说起刚开始学网络编程那会儿大家写的第一个 TCP 服务器多半是“阻塞 accept 循环 recv”。伪代码长这样while (1) { int cfd accept(lstfd, ...); handle(cfd); // 读取请求返回响应 }这个程序只能同时服务一个连接只要 handle 里 recv 一阻塞后面的客户端全部排队。于是很快有人改成 fork 或者 thread-per-connectionwhile (1) { int cfd accept(lstfd, ...); pthread_create(tid, NULL, handle, cfd); }每个连接一个线程确实解决问题了。但当你做过一段时间服务端开发就会发现这种模型有几个非常实在的痛点线程多了以后上下文切换开销上去了每个线程栈动辄几 MB内存不够用更麻烦的是如果某个连接一直没有数据它的线程会白白占着资源阻塞在 recv 上。我刚工作那几年见过一个部署在某银行的接入服务高峰期开了三千多个线程CPU 大量消耗在切线程上而大多数线程只是等数据发呆。后来接触 Reactor 模型才意识到“一连接一线程”的本质是把“空闲等待”也当成一种资源消耗。而事件驱动想解决的是让一个线程同时监听几千个 fd谁有数据就处理谁没有数据就去等下一个。这才是服务器能扛高并发的关键思路转变。1.2 非阻塞 IO 与“轮询”的进化Reactor 背后依赖的非阻塞 IO 并不是新东西但它一定要搭配一个高效的“通知机制”才有意义。早期用 select/poll虽然也是单线程监听多个 fd但每次调用都需要线性遍历所有 fd连接数一上去性能就崩。一直到 Linux 下的 epoll、BSD 系的 kqueue 出现才真正把“就绪通知”变成了 O(1) 级别的回调触发。说白了epoll 像什么呢像你在商场里等十个朋友select 的做法是每隔几秒把商场广播一遍“张三走了吗李四走了吗”一个一个问。epoll 的做法是让每个朋友出来时主动按一下你手机上的铃声谁按你才看谁。这件事的底层实现是内核帮你维护了一棵红黑树和一张就绪链表当有 fd 上的事件发生时内核直接把对应的 file descriptor 放到就绪链表里你调用 epoll_wait 时拿到的就是“已经活跃的人”的名单。本文要聊的 Reactor 就是建立在 epoll 这类多路复用器之上的编程范式。为了让你彻底看明白我打算用 Python 标准库里的 selectors 模块手写一个极简 Reactor——它底层会自动选择 epoll代码又比 C 短很多。然后用这个框架实现一个有意思的实验“自发自送 TCP 服务器”。所谓自发自送就是服务器进程里同时扮演客户端和服务端自己连自己自己发数据自己再把数据收回来读一遍。这么一个小玩具能把 Reactor 的事件分发、注册、回调、非阻塞 connect 这些核心机制全部串起来比单纯的 echo server 更能说明问题。2. Reactor 的五个角色一张数据流转图走完一次 recv2.1 Handler、Demultiplexer、Event Loop 到底各管什么很多文章讲 Reactor 都会提“五个角色”但真正写代码时很多人分不清楚这些角的边界。我按自己实现时的理解重新梳理一遍角色对应代码实体职责事件源Event Sourcefd、socket 对象内核可感知的 IO 对象可能是 listen fd也可能是普通连接事件多路分发器Demultiplexerselectors.DefaultSelector() 或 epoll 实例等事件、收集就绪 fd这是内核级的阻塞等待事件循环Event LoopEventLoop.run() 里的 while 循环不断调用多路分发器拿到就绪事件后分发给回调事件处理器Handleron_accept / on_recv / on_write具体的业务回调比如 accept 一个连接、读取数据反应器Reactor 本身EventLoop 类 Channel 注册表将以上角色串成一个自动运行的体系含注册、删除、修改事件的 API上面表格里最容易被忽略的是“反应器”和“事件循环”的区别。事件循环只是个死循环它负责“转”反应器还承担了“管理”的职责谁注册进来、注册的是哪个 fd、关心什么事件、对应哪个回调这些元信息都在反应器里维护。我的 EventLoop 类实际上就是反应器和事件循环合二为一这是小型实现里很常见的做法。2.2 一次 recv 事件从发生到业务函数被调用的完整链路假设客户端已经建立连接并发送了 8 个字节的数据那么在 Reactor 体系里这些字节从内核到用户态业务代码会走这样一条路线数据到达网卡内核把数据放入 socket 对应的接收缓冲区。内核检测到该 socket 可读把对应的 fd 加入 epoll 的就绪链表。站在事件循环里的线程此时可能正阻塞在selector.select()上这个调用立刻返回给出“哪个 fd 哪个事件发生了”的信息。事件循环读取注册表找到这个 fd 对应的回调函数。循环调用回调把事件掩码传进去比如EVENT_READ。回调里执行recv()注意此时是非阻塞的而且一定读得到数据因为事件通知已经说明“读不会阻塞”。回调读完数据后返回事件循环回到select()继续等下一个事件。这条链路里最反直觉的一点是回调函数在执行时是同一条线程。也就是说如果你的回调里写了sleep(10)整个事件循环会卡死 10 秒所有其他 fd 的事件都会堆积。第一次用 Reactor 写业务逻辑的人十有八九在这上面栽过跟头。避免方法只有两个要么回调里只做快操作要么把重活丢到线程池让业务线程处理处理完再通过线程安全的队列唤醒事件循环。2.3 注册表里的“兴趣事件”为什么总是带掩码注册一个 fd 到 Reactor 时要同时声明“我关心哪些事件”。Linux 的 epoll 里常见的事件掩码是EPOLLIN可读对应 selectors 的EVENT_READEPOLLOUT可写对应 selectors 的EVENT_WRITEEPOLLERR/EPOLLHUP错误、对端挂断EPOLLRDHUP对端关闭写端常用于 HTTP/2 快速感知对端关闭掩码的含义是“兴趣”不是“状态”。比如你注册EPOLLOUT并不代表 fd 一定可写而是告诉内核“当它可以写的时候请通知我”。这跟阻塞编程里“write 一定成功”的直觉完全不同。你后面看自发自送代码时会看到 non-blocking connect 需要先注册EVENT_WRITE原因就在这里connect 在非阻塞 fd 上发出后连接建立完成这个事件在 epoll 里就是“可写事件”。3. 手写轻量 ReactorEventLoop、Channel、回调三件套3.1 最精简的框架骨架代码Python 的 selectors 模块把底层的 epoll 封装得比 C 舒服太多但我仍然要把框架写成类方便你以后迁移到 C 或者 Java NIO 理解。首先是最核心的 EventLoop 类import socket import selectors class EventLoop: def __init__(self): self.sel selectors.DefaultSelector() def add_reader(self, sock, callback): self.sel.register(sock, selectors.EVENT_READ, lambda mask: callback(sock, mask)) def add_writer(self, sock, callback): self.sel.register(sock, selectors.EVENT_WRITE, lambda mask: callback(sock, mask)) def modify(self, sock, events, callback): self.sel.modify(sock, events, lambda mask: callback(sock, mask)) def remove(self, sock): try: self.sel.unregister(sock) except Exception: pass sock.close() def run(self): while True: events self.sel.select(timeout1) for key, mask in events: key.data(mask)这段代码非常短但已经具备 Reactor 的基本能力。注意add_reader和add_writer里传入的是一个闭包它把sock和callback绑死在了一起。这样事件循环不用关心 fd 映射的事只需要拿到就绪事件后执行key.data(mask)。modify方法很重要因为同一个 fd 在事件不同阶段关心的内容不同。比如客户端 socket 刚 connect 成功时需要监听可写发送完数据后要改成监听可读。如果没有modify就得 remove 再 add中间会漏掉事件。3.2 选择器 timeout 参数该不该给给多少上面run()里我用的是select(timeout1)这其实是个很值得解释的小设计。如果你把 timeout 设为 None线程会一直阻塞直到有事件发生。这在纯网络服务器里没问题但如果你希望这个服务器还能处理定时任务、或者能优雅退出就必须给一个非 None 的 timeout。常见做法是给 1 秒这样每隔 1 秒run()会醒来一次可以检查关机标志、执行定时回调、清理超时连接。我当年第一次写 Reactor 时给了 30 秒结果服务关停时要等 30 秒才响应停止命令非常痛苦。后来改成 1 秒CPU 开销几乎可以忽略因为 Linux 内核的定时精度远高于此。如果是毫秒级超时需求建议只在你自己的循环里维护最小堆定时器而不是把 epoll timeout 设得太小。3.3 accept 流程监听 fd 和业务连接 fd 是两件事有了 EventLoop一个最基础的 acceptor 代码是这样def on_accept(sock, mask): conn, addr sock.accept() conn.setblocking(False) loop.add_reader(conn, on_recv)很多新手会犯一个错误监听 socket 和 accept 出来的连接 socket 混用一个回调。其实监听 socket 只负责暴露新连接的“入口”它的 fd 永远是同一个而 accpet 出来的 socket 才是实际通信通道。在 epoll 里它们要有两个独立的注册项监听 fd 注册EVENT_READ连接 fd 也注册EVENT_READ但回调完全不一样。事件循环分发的依据是注册时绑定的回调所以千万别在回调里再判断“我到底是哪个 socket”。还要注意conn.setblocking(False)这一步。默认情况下accept()返回的 socket 是继承监听 socket 的阻塞属性的如果你不手动改非阻塞那么后面在回调里执行recv()时会因为读不到数据而阻塞整个事件循环。这个坑在 C 语言的 epoll 实现里也一样区别只是 Python 会直接抛BlockingIOError告诉你。4. 自发自送 TCP 服务器的设计自己连自己把闭环跑起来4.1 为什么要做这样一个反直觉的实验平时调试网络服务最麻烦的就是准备客户端。要么拿nc敲命令要么写一个临时脚本而且如果客户端和服务端不在同一个网络环境还要同步部署很烦。“自发自送”的意思是在一段 TCP 服务器代码内同时完成“创建监听”“主动 connect 自己”“向自己发送数据”“通过自己的接受回调读取数据”四个动作。这样做的好处有两个第一个好处是闭环可验证。你不需要外部工具程序一启动自己发自己收如果能正常打印出数据说明 Reactor 的事件注册、回调分发、非阻塞 read/write 全部正确。第二个好处是能暴露客户端视角的坑。平时写服务器代码时很少有人认真走一遍非阻塞 connect 的过程而这个实验强制你处理EINPROGRESS、EPOLLOUT这些东西。等服务端开发时遇到类似场景你心里就有谱了。4.2 完整代码一场自发自送的事件之旅下面是完整的 Python 代码核心结构就是上一节的 EventLoop 加上自发自送逻辑。代码未做极端优化重点在清晰。import socket import selectors import sys class EventLoop: def __init__(self): self.sel selectors.DefaultSelector() def add_reader(self, sock, callback): self.sel.register(sock, selectors.EVENT_READ, lambda mask: callback(sock, mask)) def add_writer(self, sock, callback): self.sel.register(sock, selectors.EVENT_WRITE, lambda mask: callback(sock, mask)) def modify(self, sock, events, callback): self.sel.modify(sock, events, lambda mask: callback(sock, mask)) def remove(self, sock): try: self.sel.unregister(sock) except Exception: pass print(f[loop] close fd {sock.fileno()}) sock.close() def run(self): try: while True: events self.sel.select(timeout1) for key, mask in events: key.data(mask) except KeyboardInterrupt: pass PORT 54321 loop EventLoop() def on_accept(sock, mask): conn, addr sock.accept() conn.setblocking(False) print(f[server] accept new connection from {addr}, fd{conn.fileno()}) loop.add_reader(conn, on_server_recv) def on_server_recv(sock, mask): try: data sock.recv(1024) except BlockingIOError: return if data: msg data.decode(errorsreplace).strip() print(f[server] received: {msg}) sock.sendall(becho: data) else: print(f[server] peer closed, removing fd {sock.fileno()}) loop.remove(sock) def on_client_write(sock, mask): # 非阻塞connect完成socket变为可写此时可以发送数据 err sock.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR) if err ! 0: print(f[client] connect failed: {err}) loop.remove(sock) return print(f[client] connected to server, sending hello self) try: sock.sendall(bhello self) except BlockingIOError: pass # 发送完毕改为等待回显 loop.modify(sock, selectors.EVENT_READ, on_client_recv) def on_client_recv(sock, mask): try: data sock.recv(1024) except BlockingIOError: return if data: print(f[client] received echo: {data.decode(errorsreplace).strip()}) print([client] round-trip OK, shut down) loop.remove(sock) else: loop.remove(sock) def main(): global loop # 1. 创建服务端监听socket server socket.socket(socket.AF_INET, socket.SOCK_STREAM) server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server.bind((127.0.0.1, PORT)) server.listen(16) server.setblocking(False) print(f[server] listening on 127.0.0.1:{PORT}, fd{server.fileno()}) loop.add_reader(server, on_accept) # 2. 创建客户端socket非阻塞connect到服务端 client socket.socket(socket.AF_INET, socket.SOCK_STREAM) client.setblocking(False) rc client.connect_ex((127.0.0.1, PORT)) print(f[client] connect_ex returned {rc}) if rc 0: # 意外情况连接立即建立成功了直接发数据 loop.add_writer(client, on_client_write) else: # 现实中通常返回 errno 115EINPROGRESS loop.add_writer(client, on_client_write) print([loop] event loop starting...) loop.run() if __name__ __main__: main()这个代码跑起来后你会看到类似这样的输出[server] listening on 127.0.0.1:54321, fd3 [client] connect_ex returned 115 [loop] event loop starting... [server] accept new connection from (127.0.0.1, 56789), fd4 [client] connected to server, sending hello self [server] received: hello self [client] received echo: echo: hello self [client] round-trip OK, shut down仔细观察你会发现 accept 和 connect 完成这两个事件的先后顺序并不固定这取决于操作系统调度。但无论如何它们都由同一条事件循环线程处理这就是 Reactor 最迷人也最需要习惯的地方。4.3 把非阻塞 connect 的几个状态捋清楚上面代码里connect_ex返回 115这是 EINPROGRESS也就是“连接已经开始建立但还没有完成”。在阻塞模式下connect 会一直等到三次握手完成才返回在非阻塞模式下connect 会立即返回 EINPROGRESS之后内核继续握手。你需要等到 socket“可写”时再用getsockopt(SOL_SOCKET, SO_ERROR)拿一次连接结果判断到底是成功还是失败。这一步特别容易出错错误做法是在可写事件回调里直接send()。理论上如果连接失败也会触发可写事件直接 send 会覆盖真正的错误。所以正确顺序是connect_ex 返回 EINPROGRESS。把 socket 注册到 Reactor 的写事件。可写事件触发先查 SO_ERROR。检查 err 为 0 才表示连接成功随后才能 send。还有一个小细节Windows 上非阻塞 connect 的行为和 Linux 略有差异处理错误时要考虑WSAEWOULDBLOCK之类的语义。我这边的实验环境是 Linux如果你在 mac/Windows 跑多打几个日志观察一下。5. 跑通之后的三个坑EINPROGRESS、部分写入、事件滞留5.1 坑一sendall 在非阻塞 socket 上并不保证“全部发送”代码里我用了sock.sendall(bhello self)在数据量小的时候它没问题。原因很简单内核发送缓冲区足够大一次 send 就全部放进缓冲区了。但如果数据量大到超过发送缓冲区非阻塞模式下 send 只会发送能容纳的部分返回实际发送字节数。sendall在非阻塞 socket 上内部会循环调用 send直到发完或者遇到 BlockingIOError而遇到 BlockingIOError 时它可能抛出异常也可能丢数据具体表现形式取决于 Python 版本和底层行为。真正工程化的做法是给每个连接配一个出站缓冲区。你要发送的数据先 append 到缓冲区然后尝试 flush如果没发完就注册写事件等下一次可写事件再继续 flush。这个过程在事件循环里非常常见Netty、libevent 都是这么处理的。对于本文的演示代码数据量小所以无所谓但你千万别在真实服务器里直接依赖非阻塞的 sendall。5.2 坑二事件滞留与修改事件的时机在自发自送代码里客户端 socket 一开始注册的是写事件当连接完成、发送完数据之后我立刻用loop.modify把监听事件从EVENT_WRITE改成EVENT_READ。如果不做这个修改会发生什么答案是可写事件会持续触发因为一个正常建立的 socket 永远处于可写状态。换在 epoll 的 level-trigger 模式下你的回调会被疯狂调用CPU 直接跑满。这个“事件滞留”的现象在注册了写事件但暂时不需要发送数据时经常遇到。所以 Reactor 里有个不成文的规则不需要写事件时立刻改回只监读。有些框架封装了send()方法自动管理这个状态本质逻辑就是这样。另外如果用的是 epoll 的 edge-trigger 模式事件只触发一次反而容易漏发你还得手动再注册一次。初学者用 level-trigger 更能理解逻辑生产环境则要小心两者取舍。5.3 坑三服务端回调里直接回显缺少应用层缓冲你可能会想服务端收到数据后直接sendall回去不是很简单吗是的回显流程在数据量小的时候很简单。但真实场景里你需要考虑半包和粘包的问题。比如客户端分两次发送hello和world服务端可能一次 recv 读取到hello world也可能只读到hello甚至只读到hel。TCP 是流协议它不保证消息边界。Reactor 不解决拆包问题它只保证“有数据读的时候回调你”。所以生产代码里必须有一个应用层缓冲区把收到的数据暂存起来再按你设计的协议格式去判断有没有完整包。常见方案有固定长度、特殊分隔符、TLV 格式、或长度前缀。我见过很多自研网络框架挂在“粘包”这个最简单的问题上本质上就是没理解 Reactor 只负责搬运协议解析得自己做。这张表格可以帮你快速对照常见问题现象原因解决connect_ex 返回 EINPROGRESS 后一直没回调忘了注册写事件注册EVENT_WRITE等待连接完成可写事件疯狂触发level-trigger 模式发送完毕后修改监听事件为EVENT_READrecv 总是读到半包TCP 流式传输添加应用层缓冲区做拆包回调里 sleep 后所有 fd 都不响应单线程事件循环被阻塞回调只做轻操作重活放线程池对端断开后回调反复触发未正确处理 EOFrecv 返回空字节时要移除 fd5.4 如何验证你的“自发自送”确实用上了 Reactor 的全部能力代码跑通只是第一步。我建议你做两个改造实验来加深对这套机制的理解。第一个实验在on_accept里故意加一个print并马上返回观察什么事件都不会被处理因为新连接并没有注册到事件循环。这说明 Reactor 里“注册”是回调被调用的前置条件。第二个实验注释掉loop.modify(client, ...)那一行重新运行。你会看到客户端可写回调被不断触发服务端会反复收到hello self并反复回显。这一现象比任何文字都更能说明“事件兴趣管理”的重要性。这两个实验做完你对 Reactor 的基本心智模型就建立起来了。6. 从玩具到工程如果你要把它扩展成真正的服务器6.1 核心框架可以从单线程扩到多线程模型我前面写的 EventLoop 是单线程的。生产级服务器通常不会真的用单线程硬扛所有连接、CPU 密集任务和网络 IO。常见方案是“主从 Reactor”主线程只 accept 新连接然后把连接分发给若干个工作线程每个工作线程有自己的 EventLoop处理自己的连接集合。这样做的好处是 accept 和业务处理隔离多个事件循环可以并行跑在不同 CPU 上。扩展到多线程模型时最麻烦的不是业务代码而是线程间的“交接”。比如主线程 accept 得到新 socket 后要把它注册到某个 worker 的 EventLoop 里而这个 worker 正阻塞在 select 上。如何唤醒它Linux 下通用的做法是给每个 event loop 准备一个eventfd需要注册新 fd 时往 eventfd 写一个字节select 立刻返回。这个细节在任何高并发服务器框架里都很关键。6.2 定时器、信号处理和优雅退出只有当你不满足于“永远 while True”才会意识到 Reactor 还需要管理超时和退出。比如一个 TCP 连接长时间不发数据你可能要踢掉它服务收到 SIGTERM 要关闭所有连接、清理资源。单线程的 EventLoop 里实现这些的方法就是给 select 加 timeout在每次醒来时检查定时器堆和关闭标志。考虑到你看到的框架源码里一定有run()/stop()方法你现在应该能理解它们无非就是在事件循环里维护两个集合一个是 fd 到回调的映射一个是时间到回调的映射。科目再多核心还是那张就绪事件表。6.3 用不用框架你都得懂这套内功现在市面上有 Netty、libevent、Asio 这些成熟框架你可能会觉得手写 Reactor 没必要。但我在维护遗留系统时多次发现出问题的往往是框架使用不当而不是框架本身。比如有人用 Netty 在 channelRead 里面做数据库同步查询直接卡死 event loop有人用 libevent 却没注意写事件注册导致的忙轮询。这些问题的根源都是对本文涉及的最小模型理解不透。所以我还是建议每一个立志做服务端开发的人至少完整手写一次 Reactor。不需要复杂一个 EventLoop 加两个回调就够。当你亲手看到“accept”和“connect 完成”在同一条线程里交替发生当你调试过连接建立失败却没触发任何回调的诡异问题你对高性能 IO 的理解会上一个台阶。最后再分享一个调这类代码的小技巧日志里永远带上 fd 编号和事件掩码。你在add_reader和回调入口各写一行 debug就能很清楚地看到某个 fd 的状态迁移。比在那儿脑补事件循环转得快多了。 
长河君 · 2026/10/19 4:49:35 · 阅读 1.2k 手写极简Reactor：用事件驱动与epoll实现自发自送TCP服务器
1. 为什么“事件驱动”能替代“一连接一线程”这个老思路1.1 从最简单的 accept 阻塞说起刚开始学网络编程那会儿大家写的第一个 TCP 服务器多半是“阻塞 accept 循环 recv”。伪代码长这样while (1) { int cfd accept(lstfd, ...); handle(cfd); // 读取请求返回响应 }这个程序只能同时服务一个连接只要 handle 里 recv 一阻塞后面的客户端全部排队。于是很快有人改成 fork 或者 thread-per-connectionwhile (1) { int cfd accept(lstfd, ...); pthread_create(tid, NULL, handle, cfd); }每个连接一个线程确实解决问题了。但当你做过一段时间服务端开发就会发现这种模型有几个非常实在的痛点线程多了以后上下文切换开销上去了每个线程栈动辄几 MB内存不够用更麻烦的是如果某个连接一直没有数据它的线程会白白占着资源阻塞在 recv 上。我刚工作那几年见过一个部署在某银行的接入服务高峰期开了三千多个线程CPU 大量消耗在切线程上而大多数线程只是等数据发呆。后来接触 Reactor 模型才意识到“一连接一线程”的本质是把“空闲等待”也当成一种资源消耗。而事件驱动想解决的是让一个线程同时监听几千个 fd谁有数据就处理谁没有数据就去等下一个。这才是服务器能扛高并发的关键思路转变。1.2 非阻塞 IO 与“轮询”的进化Reactor 背后依赖的非阻塞 IO 并不是新东西但它一定要搭配一个高效的“通知机制”才有意义。早期用 select/poll虽然也是单线程监听多个 fd但每次调用都需要线性遍历所有 fd连接数一上去性能就崩。一直到 Linux 下的 epoll、BSD 系的 kqueue 出现才真正把“就绪通知”变成了 O(1) 级别的回调触发。说白了epoll 像什么呢像你在商场里等十个朋友select 的做法是每隔几秒把商场广播一遍“张三走了吗李四走了吗”一个一个问。epoll 的做法是让每个朋友出来时主动按一下你手机上的铃声谁按你才看谁。这件事的底层实现是内核帮你维护了一棵红黑树和一张就绪链表当有 fd 上的事件发生时内核直接把对应的 file descriptor 放到就绪链表里你调用 epoll_wait 时拿到的就是“已经活跃的人”的名单。本文要聊的 Reactor 就是建立在 epoll 这类多路复用器之上的编程范式。为了让你彻底看明白我打算用 Python 标准库里的 selectors 模块手写一个极简 Reactor——它底层会自动选择 epoll代码又比 C 短很多。然后用这个框架实现一个有意思的实验“自发自送 TCP 服务器”。所谓自发自送就是服务器进程里同时扮演客户端和服务端自己连自己自己发数据自己再把数据收回来读一遍。这么一个小玩具能把 Reactor 的事件分发、注册、回调、非阻塞 connect 这些核心机制全部串起来比单纯的 echo server 更能说明问题。2. Reactor 的五个角色一张数据流转图走完一次 recv2.1 Handler、Demultiplexer、Event Loop 到底各管什么很多文章讲 Reactor 都会提“五个角色”但真正写代码时很多人分不清楚这些角的边界。我按自己实现时的理解重新梳理一遍角色对应代码实体职责事件源Event Sourcefd、socket 对象内核可感知的 IO 对象可能是 listen fd也可能是普通连接事件多路分发器Demultiplexerselectors.DefaultSelector() 或 epoll 实例等事件、收集就绪 fd这是内核级的阻塞等待事件循环Event LoopEventLoop.run() 里的 while 循环不断调用多路分发器拿到就绪事件后分发给回调事件处理器Handleron_accept / on_recv / on_write具体的业务回调比如 accept 一个连接、读取数据反应器Reactor 本身EventLoop 类 Channel 注册表将以上角色串成一个自动运行的体系含注册、删除、修改事件的 API上面表格里最容易被忽略的是“反应器”和“事件循环”的区别。事件循环只是个死循环它负责“转”反应器还承担了“管理”的职责谁注册进来、注册的是哪个 fd、关心什么事件、对应哪个回调这些元信息都在反应器里维护。我的 EventLoop 类实际上就是反应器和事件循环合二为一这是小型实现里很常见的做法。2.2 一次 recv 事件从发生到业务函数被调用的完整链路假设客户端已经建立连接并发送了 8 个字节的数据那么在 Reactor 体系里这些字节从内核到用户态业务代码会走这样一条路线数据到达网卡内核把数据放入 socket 对应的接收缓冲区。内核检测到该 socket 可读把对应的 fd 加入 epoll 的就绪链表。站在事件循环里的线程此时可能正阻塞在selector.select()上这个调用立刻返回给出“哪个 fd 哪个事件发生了”的信息。事件循环读取注册表找到这个 fd 对应的回调函数。循环调用回调把事件掩码传进去比如EVENT_READ。回调里执行recv()注意此时是非阻塞的而且一定读得到数据因为事件通知已经说明“读不会阻塞”。回调读完数据后返回事件循环回到select()继续等下一个事件。这条链路里最反直觉的一点是回调函数在执行时是同一条线程。也就是说如果你的回调里写了sleep(10)整个事件循环会卡死 10 秒所有其他 fd 的事件都会堆积。第一次用 Reactor 写业务逻辑的人十有八九在这上面栽过跟头。避免方法只有两个要么回调里只做快操作要么把重活丢到线程池让业务线程处理处理完再通过线程安全的队列唤醒事件循环。2.3 注册表里的“兴趣事件”为什么总是带掩码注册一个 fd 到 Reactor 时要同时声明“我关心哪些事件”。Linux 的 epoll 里常见的事件掩码是EPOLLIN可读对应 selectors 的EVENT_READEPOLLOUT可写对应 selectors 的EVENT_WRITEEPOLLERR/EPOLLHUP错误、对端挂断EPOLLRDHUP对端关闭写端常用于 HTTP/2 快速感知对端关闭掩码的含义是“兴趣”不是“状态”。比如你注册EPOLLOUT并不代表 fd 一定可写而是告诉内核“当它可以写的时候请通知我”。这跟阻塞编程里“write 一定成功”的直觉完全不同。你后面看自发自送代码时会看到 non-blocking connect 需要先注册EVENT_WRITE原因就在这里connect 在非阻塞 fd 上发出后连接建立完成这个事件在 epoll 里就是“可写事件”。3. 手写轻量 ReactorEventLoop、Channel、回调三件套3.1 最精简的框架骨架代码Python 的 selectors 模块把底层的 epoll 封装得比 C 舒服太多但我仍然要把框架写成类方便你以后迁移到 C 或者 Java NIO 理解。首先是最核心的 EventLoop 类import socket import selectors class EventLoop: def __init__(self): self.sel selectors.DefaultSelector() def add_reader(self, sock, callback): self.sel.register(sock, selectors.EVENT_READ, lambda mask: callback(sock, mask)) def add_writer(self, sock, callback): self.sel.register(sock, selectors.EVENT_WRITE, lambda mask: callback(sock, mask)) def modify(self, sock, events, callback): self.sel.modify(sock, events, lambda mask: callback(sock, mask)) def remove(self, sock): try: self.sel.unregister(sock) except Exception: pass sock.close() def run(self): while True: events self.sel.select(timeout1) for key, mask in events: key.data(mask)这段代码非常短但已经具备 Reactor 的基本能力。注意add_reader和add_writer里传入的是一个闭包它把sock和callback绑死在了一起。这样事件循环不用关心 fd 映射的事只需要拿到就绪事件后执行key.data(mask)。modify方法很重要因为同一个 fd 在事件不同阶段关心的内容不同。比如客户端 socket 刚 connect 成功时需要监听可写发送完数据后要改成监听可读。如果没有modify就得 remove 再 add中间会漏掉事件。3.2 选择器 timeout 参数该不该给给多少上面run()里我用的是select(timeout1)这其实是个很值得解释的小设计。如果你把 timeout 设为 None线程会一直阻塞直到有事件发生。这在纯网络服务器里没问题但如果你希望这个服务器还能处理定时任务、或者能优雅退出就必须给一个非 None 的 timeout。常见做法是给 1 秒这样每隔 1 秒run()会醒来一次可以检查关机标志、执行定时回调、清理超时连接。我当年第一次写 Reactor 时给了 30 秒结果服务关停时要等 30 秒才响应停止命令非常痛苦。后来改成 1 秒CPU 开销几乎可以忽略因为 Linux 内核的定时精度远高于此。如果是毫秒级超时需求建议只在你自己的循环里维护最小堆定时器而不是把 epoll timeout 设得太小。3.3 accept 流程监听 fd 和业务连接 fd 是两件事有了 EventLoop一个最基础的 acceptor 代码是这样def on_accept(sock, mask): conn, addr sock.accept() conn.setblocking(False) loop.add_reader(conn, on_recv)很多新手会犯一个错误监听 socket 和 accept 出来的连接 socket 混用一个回调。其实监听 socket 只负责暴露新连接的“入口”它的 fd 永远是同一个而 accpet 出来的 socket 才是实际通信通道。在 epoll 里它们要有两个独立的注册项监听 fd 注册EVENT_READ连接 fd 也注册EVENT_READ但回调完全不一样。事件循环分发的依据是注册时绑定的回调所以千万别在回调里再判断“我到底是哪个 socket”。还要注意conn.setblocking(False)这一步。默认情况下accept()返回的 socket 是继承监听 socket 的阻塞属性的如果你不手动改非阻塞那么后面在回调里执行recv()时会因为读不到数据而阻塞整个事件循环。这个坑在 C 语言的 epoll 实现里也一样区别只是 Python 会直接抛BlockingIOError告诉你。4. 自发自送 TCP 服务器的设计自己连自己把闭环跑起来4.1 为什么要做这样一个反直觉的实验平时调试网络服务最麻烦的就是准备客户端。要么拿nc敲命令要么写一个临时脚本而且如果客户端和服务端不在同一个网络环境还要同步部署很烦。“自发自送”的意思是在一段 TCP 服务器代码内同时完成“创建监听”“主动 connect 自己”“向自己发送数据”“通过自己的接受回调读取数据”四个动作。这样做的好处有两个第一个好处是闭环可验证。你不需要外部工具程序一启动自己发自己收如果能正常打印出数据说明 Reactor 的事件注册、回调分发、非阻塞 read/write 全部正确。第二个好处是能暴露客户端视角的坑。平时写服务器代码时很少有人认真走一遍非阻塞 connect 的过程而这个实验强制你处理EINPROGRESS、EPOLLOUT这些东西。等服务端开发时遇到类似场景你心里就有谱了。4.2 完整代码一场自发自送的事件之旅下面是完整的 Python 代码核心结构就是上一节的 EventLoop 加上自发自送逻辑。代码未做极端优化重点在清晰。import socket import selectors import sys class EventLoop: def __init__(self): self.sel selectors.DefaultSelector() def add_reader(self, sock, callback): self.sel.register(sock, selectors.EVENT_READ, lambda mask: callback(sock, mask)) def add_writer(self, sock, callback): self.sel.register(sock, selectors.EVENT_WRITE, lambda mask: callback(sock, mask)) def modify(self, sock, events, callback): self.sel.modify(sock, events, lambda mask: callback(sock, mask)) def remove(self, sock): try: self.sel.unregister(sock) except Exception: pass print(f[loop] close fd {sock.fileno()}) sock.close() def run(self): try: while True: events self.sel.select(timeout1) for key, mask in events: key.data(mask) except KeyboardInterrupt: pass PORT 54321 loop EventLoop() def on_accept(sock, mask): conn, addr sock.accept() conn.setblocking(False) print(f[server] accept new connection from {addr}, fd{conn.fileno()}) loop.add_reader(conn, on_server_recv) def on_server_recv(sock, mask): try: data sock.recv(1024) except BlockingIOError: return if data: msg data.decode(errorsreplace).strip() print(f[server] received: {msg}) sock.sendall(becho: data) else: print(f[server] peer closed, removing fd {sock.fileno()}) loop.remove(sock) def on_client_write(sock, mask): # 非阻塞connect完成socket变为可写此时可以发送数据 err sock.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR) if err ! 0: print(f[client] connect failed: {err}) loop.remove(sock) return print(f[client] connected to server, sending hello self) try: sock.sendall(bhello self) except BlockingIOError: pass # 发送完毕改为等待回显 loop.modify(sock, selectors.EVENT_READ, on_client_recv) def on_client_recv(sock, mask): try: data sock.recv(1024) except BlockingIOError: return if data: print(f[client] received echo: {data.decode(errorsreplace).strip()}) print([client] round-trip OK, shut down) loop.remove(sock) else: loop.remove(sock) def main(): global loop # 1. 创建服务端监听socket server socket.socket(socket.AF_INET, socket.SOCK_STREAM) server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server.bind((127.0.0.1, PORT)) server.listen(16) server.setblocking(False) print(f[server] listening on 127.0.0.1:{PORT}, fd{server.fileno()}) loop.add_reader(server, on_accept) # 2. 创建客户端socket非阻塞connect到服务端 client socket.socket(socket.AF_INET, socket.SOCK_STREAM) client.setblocking(False) rc client.connect_ex((127.0.0.1, PORT)) print(f[client] connect_ex returned {rc}) if rc 0: # 意外情况连接立即建立成功了直接发数据 loop.add_writer(client, on_client_write) else: # 现实中通常返回 errno 115EINPROGRESS loop.add_writer(client, on_client_write) print([loop] event loop starting...) loop.run() if __name__ __main__: main()这个代码跑起来后你会看到类似这样的输出[server] listening on 127.0.0.1:54321, fd3 [client] connect_ex returned 115 [loop] event loop starting... [server] accept new connection from (127.0.0.1, 56789), fd4 [client] connected to server, sending hello self [server] received: hello self [client] received echo: echo: hello self [client] round-trip OK, shut down仔细观察你会发现 accept 和 connect 完成这两个事件的先后顺序并不固定这取决于操作系统调度。但无论如何它们都由同一条事件循环线程处理这就是 Reactor 最迷人也最需要习惯的地方。4.3 把非阻塞 connect 的几个状态捋清楚上面代码里connect_ex返回 115这是 EINPROGRESS也就是“连接已经开始建立但还没有完成”。在阻塞模式下connect 会一直等到三次握手完成才返回在非阻塞模式下connect 会立即返回 EINPROGRESS之后内核继续握手。你需要等到 socket“可写”时再用getsockopt(SOL_SOCKET, SO_ERROR)拿一次连接结果判断到底是成功还是失败。这一步特别容易出错错误做法是在可写事件回调里直接send()。理论上如果连接失败也会触发可写事件直接 send 会覆盖真正的错误。所以正确顺序是connect_ex 返回 EINPROGRESS。把 socket 注册到 Reactor 的写事件。可写事件触发先查 SO_ERROR。检查 err 为 0 才表示连接成功随后才能 send。还有一个小细节Windows 上非阻塞 connect 的行为和 Linux 略有差异处理错误时要考虑WSAEWOULDBLOCK之类的语义。我这边的实验环境是 Linux如果你在 mac/Windows 跑多打几个日志观察一下。5. 跑通之后的三个坑EINPROGRESS、部分写入、事件滞留5.1 坑一sendall 在非阻塞 socket 上并不保证“全部发送”代码里我用了sock.sendall(bhello self)在数据量小的时候它没问题。原因很简单内核发送缓冲区足够大一次 send 就全部放进缓冲区了。但如果数据量大到超过发送缓冲区非阻塞模式下 send 只会发送能容纳的部分返回实际发送字节数。sendall在非阻塞 socket 上内部会循环调用 send直到发完或者遇到 BlockingIOError而遇到 BlockingIOError 时它可能抛出异常也可能丢数据具体表现形式取决于 Python 版本和底层行为。真正工程化的做法是给每个连接配一个出站缓冲区。你要发送的数据先 append 到缓冲区然后尝试 flush如果没发完就注册写事件等下一次可写事件再继续 flush。这个过程在事件循环里非常常见Netty、libevent 都是这么处理的。对于本文的演示代码数据量小所以无所谓但你千万别在真实服务器里直接依赖非阻塞的 sendall。5.2 坑二事件滞留与修改事件的时机在自发自送代码里客户端 socket 一开始注册的是写事件当连接完成、发送完数据之后我立刻用loop.modify把监听事件从EVENT_WRITE改成EVENT_READ。如果不做这个修改会发生什么答案是可写事件会持续触发因为一个正常建立的 socket 永远处于可写状态。换在 epoll 的 level-trigger 模式下你的回调会被疯狂调用CPU 直接跑满。这个“事件滞留”的现象在注册了写事件但暂时不需要发送数据时经常遇到。所以 Reactor 里有个不成文的规则不需要写事件时立刻改回只监读。有些框架封装了send()方法自动管理这个状态本质逻辑就是这样。另外如果用的是 epoll 的 edge-trigger 模式事件只触发一次反而容易漏发你还得手动再注册一次。初学者用 level-trigger 更能理解逻辑生产环境则要小心两者取舍。5.3 坑三服务端回调里直接回显缺少应用层缓冲你可能会想服务端收到数据后直接sendall回去不是很简单吗是的回显流程在数据量小的时候很简单。但真实场景里你需要考虑半包和粘包的问题。比如客户端分两次发送hello和world服务端可能一次 recv 读取到hello world也可能只读到hello甚至只读到hel。TCP 是流协议它不保证消息边界。Reactor 不解决拆包问题它只保证“有数据读的时候回调你”。所以生产代码里必须有一个应用层缓冲区把收到的数据暂存起来再按你设计的协议格式去判断有没有完整包。常见方案有固定长度、特殊分隔符、TLV 格式、或长度前缀。我见过很多自研网络框架挂在“粘包”这个最简单的问题上本质上就是没理解 Reactor 只负责搬运协议解析得自己做。这张表格可以帮你快速对照常见问题现象原因解决connect_ex 返回 EINPROGRESS 后一直没回调忘了注册写事件注册EVENT_WRITE等待连接完成可写事件疯狂触发level-trigger 模式发送完毕后修改监听事件为EVENT_READrecv 总是读到半包TCP 流式传输添加应用层缓冲区做拆包回调里 sleep 后所有 fd 都不响应单线程事件循环被阻塞回调只做轻操作重活放线程池对端断开后回调反复触发未正确处理 EOFrecv 返回空字节时要移除 fd5.4 如何验证你的“自发自送”确实用上了 Reactor 的全部能力代码跑通只是第一步。我建议你做两个改造实验来加深对这套机制的理解。第一个实验在on_accept里故意加一个print并马上返回观察什么事件都不会被处理因为新连接并没有注册到事件循环。这说明 Reactor 里“注册”是回调被调用的前置条件。第二个实验注释掉loop.modify(client, ...)那一行重新运行。你会看到客户端可写回调被不断触发服务端会反复收到hello self并反复回显。这一现象比任何文字都更能说明“事件兴趣管理”的重要性。这两个实验做完你对 Reactor 的基本心智模型就建立起来了。6. 从玩具到工程如果你要把它扩展成真正的服务器6.1 核心框架可以从单线程扩到多线程模型我前面写的 EventLoop 是单线程的。生产级服务器通常不会真的用单线程硬扛所有连接、CPU 密集任务和网络 IO。常见方案是“主从 Reactor”主线程只 accept 新连接然后把连接分发给若干个工作线程每个工作线程有自己的 EventLoop处理自己的连接集合。这样做的好处是 accept 和业务处理隔离多个事件循环可以并行跑在不同 CPU 上。扩展到多线程模型时最麻烦的不是业务代码而是线程间的“交接”。比如主线程 accept 得到新 socket 后要把它注册到某个 worker 的 EventLoop 里而这个 worker 正阻塞在 select 上。如何唤醒它Linux 下通用的做法是给每个 event loop 准备一个eventfd需要注册新 fd 时往 eventfd 写一个字节select 立刻返回。这个细节在任何高并发服务器框架里都很关键。6.2 定时器、信号处理和优雅退出只有当你不满足于“永远 while True”才会意识到 Reactor 还需要管理超时和退出。比如一个 TCP 连接长时间不发数据你可能要踢掉它服务收到 SIGTERM 要关闭所有连接、清理资源。单线程的 EventLoop 里实现这些的方法就是给 select 加 timeout在每次醒来时检查定时器堆和关闭标志。考虑到你看到的框架源码里一定有run()/stop()方法你现在应该能理解它们无非就是在事件循环里维护两个集合一个是 fd 到回调的映射一个是时间到回调的映射。科目再多核心还是那张就绪事件表。6.3 用不用框架你都得懂这套内功现在市面上有 Netty、libevent、Asio 这些成熟框架你可能会觉得手写 Reactor 没必要。但我在维护遗留系统时多次发现出问题的往往是框架使用不当而不是框架本身。比如有人用 Netty 在 channelRead 里面做数据库同步查询直接卡死 event loop有人用 libevent 却没注意写事件注册导致的忙轮询。这些问题的根源都是对本文涉及的最小模型理解不透。所以我还是建议每一个立志做服务端开发的人至少完整手写一次 Reactor。不需要复杂一个 EventLoop 加两个回调就够。当你亲手看到“accept”和“connect 完成”在同一条线程里交替发生当你调试过连接建立失败却没触发任何回调的诡异问题你对高性能 IO 的理解会上一个台阶。最后再分享一个调这类代码的小技巧日志里永远带上 fd 编号和事件掩码。你在add_reader和回调入口各写一行 debug就能很清楚地看到某个 fd 的状态迁移。比在那儿脑补事件循环转得快多了。