首先,尝试使用eventlet
因为这是错误的工具。eventlet
主要用于使用 greenlet 形式的协程的非阻塞 I/O。它与 CPU 限制问题没有什么关系(事实上,文档eventlet.Timeout
read):
如果 try/finally 或 with-block 中的代码块从未协同产生,则无法引发超时。在 Eventlet 中,这很少会成为问题,但请注意,您不能使用此类使仅 CPU 操作超时。
这意味着,如果您有一些代码块正在执行 CPU 密集型操作,您仍然无法摆脱它,因为该代码永远不会屈服于另一个线程。这更适用于您可能有一些代码的情况,例如检查套接字是否有要读取的数据,如果没有则产生。
如果你想在一段时间后设置一个中断来中断长时间运行的计算,你可以使用signal.alarm
并设置一个SIGALRM
处理程序。您可以将这一切包装在上下文管理器中,例如:
>>> import signal
>>> from contextlib import contextmanager
>>> class TimeoutError(RuntimeError): pass
...
>>> @contextmanager
... def timeout(seconds):
... def handler(*args):
... raise TimeoutError("timed out after {} seconds".format(seconds))
... orig_handler = signal.signal(signal.SIGALRM, handler)
... signal.alarm(seconds)
... try:
... yield
... finally:
... signal.alarm(0)
... signal.signal(signal.SIGALRM, orig_handler)
...
>>> with timeout(10):
... while True: pass
...
Traceback (most recent call last):
File "<stdin>", line 2, in <module>
File "<stdin>", line 4, in handler
__main__.TimeoutError: timed out after 10 seconds
>>> with timeout(10):
... print(1 + 1)
...
2
>>> with timeout(10):
... 4 ** 5 ** 20
Traceback (most recent call last):
File "<stdin>", line 2, in <module>
File "<stdin>", line 4, in handler
__main__.TimeoutError: timed out after 10 seconds
Note:这仅适用于 *NIX 系统,尽管也有方法可以在 Windows 上执行此操作。