signal.alarm无法实现通用timeout，因其仅主线程有效、不中断纯Python计算、与多线程/异步冲突；可靠方案是threading+queue（兼容所有同步代码）或asyncio.wait_for（要求awaitable）。
为什么不能直接用 signal.alarm 实现 timeoutcontextmanager
因为 signal.alarm 只在主线程生效，且会中断系统调用（比如 time.sleep、socket.recv），但对纯 Python 计算（如死循环、大列表推导）完全无效。更麻烦的是，它和多线程、异步代码根本冲突——一旦你在子线程里设 alarm，Python 会直接抛 ValueError: signal only works in main thread。
所以自实现必须绕开信号机制，靠协程或线程协作式中断，或者依赖目标操作自身支持超时（比如 requests.get(timeout=...)）。
- 适用场景：需要包装一个「可能卡住但不响应信号」的同步阻塞调用，比如自定义网络协议解析、第三方库无 timeout 参数的函数
- 不适用场景：想给
for i in range(10**9): pass加超时——这种只能靠外部进程 kill 或改用asyncio.wait_for+ 拆成小步 yield - 关键取舍：用线程（兼容所有同步代码，但有 GIL 和资源开销）还是用 asyncio（轻量，但要求被包装函数是 awaitable）
用 threading.Thread + queue 实现最简可靠版
这是兼容性最强的方案：新开线程执行目标函数，主线程用 queue.Queue.get(timeout=...) 等结果。线程结束后自动释放，不用手动清理。
示例核心逻辑：
import threading
import queue
def timeout_contextmanager(seconds):
def decorator(func):
def wrapper(*args, **kwargs):
q = queue.Queue()
def _target():
try:
result = func(*args, **kwargs)
q.put(('success', result))
except Exception as e:
q.put(('error', e))
t = threading.Thread(target=_target, daemon=True)
t.start()
try:
status, value = q.get(timeout=seconds)
if status == 'success':
return value
raise value
except queue.Empty:
raise TimeoutError(f'Function {func.__name__} timed out after {seconds}s')
return wrapper
return decorator
- 必须设
daemon=True，否则超时时线程还在跑，程序无法退出 queue.get(timeout=...)是唯一安全的等待方式；别用t.join(seconds)—— 它不中断线程，只等结束，失去超时意义- 无法强制终止正在运行的线程（Python 没有 Thread.kill），所以超时后线程仍在后台执行，只是结果被丢弃。这对副作用操作（比如写文件、发请求）要格外小心
asyncio 版本要注意 event loop 生命周期
如果你的函数本身是 async 的，用 asyncio.wait_for 最干净。但注意：它只能在已有 event loop 的上下文中运行，不能在已关闭或未启动的 loop 里调用。
常见错误现象：RuntimeError: no running event loop 或 RuntimeWarning: coroutine ... was never awaited。
- 正确姿势：确保调用方在 async 函数内，且 event loop 正在运行（比如用
asyncio.run(main())启动） - 不要在普通函数里调用
asyncio.run(...)套一层——每次都会新建 loop，开销大且嵌套容易出错 - 参数差异：
asyncio.wait_for(coro, timeout=...)的 timeout 是 float 秒，支持小数（如0.1），而 threading 版通常只处理整数秒精度 - 性能影响：asyncio 版无额外线程开销，但要求整个调用链路都是 async，改造成本可能高于加个线程
别忽略 timeout 对异常传播的影响
超时不是“取消”，而是“放弃等待”。原函数如果在超时后仍继续执行并抛异常，这个异常不会被捕获，也不会传给上层——它就在那个孤立线程里静默消失（除非你手动 log）。
- 如果函数有副作用（如修改全局变量、写临时文件），超时后这些操作可能已完成、部分完成、或根本没开始，行为不可预测
- 想真正“取消”任务，得函数自己支持 cancellation token（比如接受一个
stop_event: threading.Event参数），timeout 只负责通知，不强制中断 - 最容易被忽略的一点：测试 timeout 路径时，别只测“刚好超时”，一定要覆盖“提前返回”和“超时后原函数仍继续运行”两种情况，否则上线后可能数据不一致
