Python中的epoll机制详解（高性能IO多路复用实战指南）

epoll vs select/poll

• select/poll：每次调用都需要传递所有要监听的文件描述符列表，内核会遍历整个列表检查状态，时间复杂度为 O(n)。
• epoll：只需在初始化时注册一次文件描述符，之后通过回调机制通知就绪事件，时间复杂度接近 O(1)，特别适合高并发场景。

Python 中如何使用 epoll？

Python 标准库中的 select 模块提供了对 epoll 的封装（仅限 Linux 系统）。

下面是一个完整的基于 epoll 的简易 TCP 服务器示例：

import socketimport selectimport osdef main():    # 创建监听 socket    server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)    server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)    server_sock.bind(('0.0.0.0', 8888))    server_sock.listen(10)    server_sock.setblocking(False)  # 设置为非阻塞    # 创建 epoll 对象    epoll = select.epoll()    # 注册监听 socket 的可读事件    epoll.register(server_sock.fileno(), select.EPOLLIN)    # 用于保存 fd 到 socket 的映射    fd_to_socket = {server_sock.fileno(): server_sock}    fd_to_data = {}  # 保存客户端接收的数据    try:        while True:            # 等待事件发生（最多等待1秒）            events = epoll.poll(1)            for fd, event in events:                sock = fd_to_socket[fd]                if sock is server_sock:                    # 新连接到来                    client_sock, addr = server_sock.accept()                    client_sock.setblocking(False)                    epoll.register(client_sock.fileno(), select.EPOLLIN)                    fd_to_socket[client_sock.fileno()] = client_sock                    fd_to_data[client_sock.fileno()] = b''                    print(f"新连接来自: {addr}")                elif event & select.EPOLLIN:                    # 客户端发送数据                    try:                        data = sock.recv(1024)                        if data:                            fd_to_data[fd] += data                            # 回显数据                            sock.send(data)                        else:                            # 客户端关闭连接                            epoll.unregister(fd)                            sock.close()                            del fd_to_socket[fd]                            del fd_to_data[fd]                    except ConnectionResetError:                        epoll.unregister(fd)                        sock.close()                        del fd_to_socket[fd]                        del fd_to_data[fd]    finally:        epoll.unregister(server_sock.fileno())        epoll.close()        server_sock.close()if __name__ == '__main__':    main()

代码解析

1. 创建非阻塞 socket：使用 setblocking(False) 避免 recv/send 阻塞主线程。
2. 注册事件：通过 epoll.register(fd, event_mask) 告诉内核我们关心哪些事件（如 EPOLLIN 表示可读）。
3. 事件循环：调用 epoll.poll() 等待事件，返回就绪的 (fd, event) 列表。
4. 处理连接与数据：区分是新连接（来自监听 socket）还是已有连接的数据到达。

注意事项

• epoll 仅在 Linux 系统可用。Windows 用户需使用 select.select() 或 asyncio。
• 务必处理异常（如客户端突然断开），否则程序可能崩溃。
• 实际项目中建议使用成熟的异步框架（如 asyncio、Tornado），它们底层已优化 epoll 使用。

总结

通过本文，你已经掌握了 Python epoll 的基本原理和实战用法。epoll 是构建高性能网络服务的核心技术之一，理解它有助于你深入掌握 Python网络编程 和 高性能IO多路复用 的精髓。

虽然手动使用 epoll 较为底层，但它是理解 asyncio 等高级异步框架的基础。建议你在学习完本教程后，尝试用 asyncio 重写上述服务器，体会高层抽象带来的便利。

关键词回顾：Python epoll、Python网络编程、epoll事件驱动、高性能IO多路复用