Python selectors.DefaultSelector 怎样管理多路 I/O 事件

当一个进程需要同时照看多个非阻塞 socket 时，最容易失控的地方不是“如何等事件”，而是“事件回来后如何找回正确的连接状态”。Python 的 selectors.DefaultSelector 把平台差异收进统一接口，应用只需要围绕注册、等待、分派和注销组织代码。

官方文档：https://docs.python.org/3/library/selectors.html

这篇文章用一个多连接监听场景拆开它的责任边界：selector 只管理就绪事件，key.data 保存业务上下文，处理函数负责读写和关闭。这样写的重点不是堆出一个更大的事件循环，而是让每个资源都有明确的清理路径。

先把三个责任边界分开

DefaultSelector 是跨平台的选择器实现入口，实际后端由 Python 根据平台选择。它不替你保存完整的连接业务状态，也不替你决定收到字节后如何解析协议。

• selector：维护哪些对象被监听，以及监听读、写还是异常相关事件。
• SelectorKey：描述某个已注册对象，包含 fileobj、fd、events 和 data。
• data：放入连接对象、缓冲区、处理函数或其他业务上下文，避免用全局字典猜测 fd 属于谁。

从安全角度看，fd 本身只是资源标识，不应该承载认证状态或协议状态。把状态放进 key.data 后，事件处理函数可以只接收一个 key，减少“连接 A 的数据写进连接 B”这类错配风险。

用 register 把非阻塞对象放进注册表

注册前要先让 socket 进入非阻塞模式，否则事件循环可能在一次 recv 或 send 中卡住，selector 的多路复用价值也就被破坏了。EVENT_READ 和 EVENT_WRITE 是位掩码，可以按需要组合。

import selectors
import socket

selector = selectors.DefaultSelector()

def register_connection(sock: socket.socket, address: tuple[str, int]) -> None:
    # 非阻塞模式让事件循环不会被单个连接拖住。
    sock.setblocking(False)
    state = {
        "address": address,
        "in_buffer": bytearray(),
        "out_buffer": bytearray(),
    }
    # 初始只关心可读事件；有待发送数据时再增加 EVENT_WRITE。
    selector.register(sock, selectors.EVENT_READ, data=state)

这里的 data 可以是字典，也可以是专门的连接对象。实际项目里如果状态字段开始变多，建议改成 dataclass，把输入缓冲、输出缓冲和关闭标记集中管理。

select 返回的是 key 和掩码，不是业务消息

selector.select(timeout) 返回一组 (key, mask)。key 告诉你是哪一个注册对象就绪，mask 告诉你这次应该处理读还是写。它返回的是 I/O 层信号，不保证一次 recv 就得到完整协议包。

def event_loop() -> None:
    # timeout=None 表示没有事件时持续等待；生产环境也可使用有限超时做巡检。
    while selector.get_map():
        ready_events = selector.select(timeout=None)
        for key, mask in ready_events:
            state = key.data
            sock = key.fileobj
            try:
                if mask & selectors.EVENT_READ:
                    read_from_socket(sock, state)
                if mask & selectors.EVENT_WRITE:
                    flush_to_socket(sock, state)
            except (ConnectionError, OSError) as exc:
                # I/O 异常时统一走清理路径，避免失效 key 留在注册表。
                close_connection(sock, reason=str(exc))

读事件中拿到空字节通常表示对端已经有序关闭；这不是“暂时没有数据”，而是连接生命周期的结束信号。写事件也不应永久注册：没有待发送数据时继续监听写事件，可能让循环不断收到可写通知却没有工作可做。

用 modify 控制写事件，避免空转

常见做法是始终监听读事件，只有输出缓冲非空时才把写事件加入掩码。发送完成后再改回只读。这样 selector 的注册表反映真实需求，而不是把所有连接都当成随时有数据要发送。

def update_interest(sock: socket.socket, state: dict) -> None:
    # 输出缓冲非空才监听写事件，避免循环因“永远可写”而空转。
    events = selectors.EVENT_READ
    if state["out_buffer"]:
        events |= selectors.EVENT_WRITE
    selector.modify(sock, events, data=state)

def flush_to_socket(sock: socket.socket, state: dict) -> None:
    # 非阻塞发送可能只写出部分数据，剩余内容必须留在缓冲区。
    if not state["out_buffer"]:
        update_interest(sock, state)
        return
    sent = sock.send(state["out_buffer"])
    del state["out_buffer"][:sent]
    update_interest(sock, state)

modify 的第三个参数仍然可以传入同一个 state，否则下一次拿到 key 时可能丢失连接上下文。对于发送失败、对端关闭或协议层主动断开，也要走同一套关闭函数。

unregister 和 close 要成对出现

selector.unregister(sock) 只负责从监听注册表移除对象，sock.close() 才负责关闭底层资源。两者不能互相替代。先注销再关闭，可以避免后续事件仍把已关闭对象当成有效连接。

def close_connection(sock: socket.socket, reason: str) -> None:
    try:
        # 对象可能已经被清理过，重复注销时把它视为幂等关闭。
        selector.unregister(sock)
    except (KeyError, ValueError):
        pass
    finally:
        # 无论注销是否成功，都释放底层 socket，避免文件描述符泄漏。
        sock.close()
        print(f"connection closed: {reason}")

清理函数最好只做一次资源释放，不在里面重新触发业务发送。若处理函数还需要写一条协议级错误，应该在关闭前完成，并且仍要考虑 send 只发送一部分或再次抛出异常。

哪些场景应该直接使用 asyncio

selectors 适合需要直接控制文件描述符、实现小型协议适配层，或为已有事件循环提供底层组件的场景。如果业务已经包含超时、任务取消、连接池、并发协程、TLS 和复杂协议状态，继续手写 selector 循环会把大量生命周期问题重新暴露出来，这时通常应优先考虑 asyncio 的流或传输抽象。

两者的关系不是“谁更快”，而是抽象层级不同：selectors 给你事件就绪和注册表，asyncio 进一步提供协程调度、任务取消和异步资源管理。先按控制需求选择，再按团队维护能力决定是否下沉到 selector。

排查多路 I/O 问题的速查表

相关问题

selectors.select 的 timeout 应该一直用 None 吗？

不一定。没有定时任务时可以阻塞等待；需要心跳、空闲连接回收或指标刷新时，使用有限超时并在空返回时处理维护任务。

为什么不能只调用 socket.close？

close 释放了 socket，但 selector 的注册表仍可能保留对应 key。正确做法是注销和关闭都放进统一的幂等清理路径。

什么时候使用 EVENT_WRITE？

只在输出缓冲确实有数据时使用，并在缓冲清空后移除写事件，否则循环可能反复收到无效的可写通知。

把 selector 看成资源登记簿，把 key.data 看成连接上下文，把 unregister 和 close 看成一组清理动作，就能把多路 I/O 的复杂度限制在几个可审查的边界内。