如何解决Python中的BlockingIOError阻塞错误_使用select或selectors模块实现非阻塞IO-Python教程-PHP中文网

blockingioerror 是非阻塞io的正常信号，表示内核暂无数据可读或缓冲区满；应通过 select 或 selectors 等机制等待就绪，而非捕获忽略或 sleep 重试。

如何解决python中的blockingioerror阻塞错误_使用select或selectors模块实现非阻塞io

BlockingIOError 本质是系统调用被中断，不是 Python 的 bug

遇到 BlockingIOError: [Errno 11] Resource temporarily unavailable，说明你对一个设为非阻塞模式的文件描述符（比如 socket、pipe）做了读/写操作，而内核当前没有数据可读或缓冲区已满。这不是异常需要“捕获并忽略”，而是非阻塞 IO 的正常反馈信号——它告诉你“现在不行，稍后再试”。关键不是绕过它，而是用正确机制等待就绪。

用 `select.select()` 手动轮询 socket 就绪状态

select.select() 是最底层、跨平台的方式，适合控制粒度要求高的场景（如单线程处理多个连接）。它不自动管理 socket 状态，需你自己维护 fd 列表和重试逻辑。

常见错误：直接在非阻塞 socket 上循环 recv() 而不先调用 select() 判断可读，导致 CPU 空转 + 频繁抛 BlockingIOError。

实操要点：

调用 select.select([read_fds], [], [], timeout) 前，确保所有 socket 已设为非阻塞（sock.setblocking(False)）
timeout 设为 0 表示非阻塞轮询；设为 None 会永久阻塞直到有事件；设为浮点数（如 0.1）是超时等待
返回的 readable 列表里包含可安全 recv() 的 socket，此时再调用不会触发 BlockingIOError
对 send()，要检查 writable 列表，且注意 send() 可能只发出部分数据，需用 sendall() 或手动处理返回值

import select
import socket
<p>sock = socket.socket()
sock.setblocking(False)
try:
sock.connect(('example.com', 80))
except BlockingIOError:
pass  # 连接开始异步进行</p><h1>等待连接完成（可写表示 connect 完成）</h1><p>readable, writable, _ = select.select([], [sock], [], 5)
if writable:
try:
sock.getpeername()  # 验证是否真的连上了
except OSError:
raise ConnectionError("connect failed")
</p>

用 `selectors.DefaultSelector` 替代手写 select 循环

selectors 模块是对 select、poll、kqueue、epoll 的统一封装，自动选择当前系统最优实现，并提供注册/注销/事件回调抽象。比裸用 select 更简洁，也更接近生产级代码习惯。

temporal-python-testing

使用 pytest、时间跳转和模拟策略测试 Temporal 工作流。涵盖单元测试、集成测试、回放测试以及本地开发服务器。

下载

容易踩的坑：

注册前必须确保 socket 是非阻塞的，否则 selector.register() 会报错
selector.select(timeout) 返回的是 (key, events) 元组列表，key.fileobj 才是 socket 对象，别直接拿 key 去 recv
EVENT_READ 和 EVENT_WRITE 不是互斥的，一个 socket 可能同时就绪读和写，需分别处理
处理完一次事件后，若还需继续监听，不用重复 register —— selector 会持续监控，除非你显式 unregister()

import selectors
import socket
<p>sel = selectors.DefaultSelector()
sock = socket.socket()
sock.setblocking(False)
try:
sock.connect(('httpbin.org', 80))
except BlockingIOError:
pass</p><h1>注册连接事件（关注可写，表示 connect 完成）</h1><p>sel.register(sock, selectors.EVENT_WRITE)</p><p>while True:
events = sel.select(timeout=5)
for key, mask in events:
if mask & selectors.EVENT_WRITE:
try:
sock.getpeername()
print("connected!")
sel.unregister(sock)  # 连上后可以 unregister 或改注册为 EVENT_READ
except OSError:
print("connection failed")
break
</p>