我写了一个简单的 TimeManager
:一个上下文管理器,在输入上下文时触发 threading.Timer
,并在退出时取消它 . 如果计时器在退出上下文之前关闭,则会引发异常:
import threading
class TimeManager(object):
def __init__(self):
self._timeout = 1
def _timeoutHandler(self):
raise Exception("Timeout!")
def __enter__(self):
self.timer = threading.Timer(self._timeout, self._timeoutHandler)
self.timer.start()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.timer.cancel()
return False
显然我无法在主线程中捕获异常,因为它属于一个单独的线程:
>>> with TimeManager() as t:
... try:
... time.sleep(5)
... except Exception:
... print "caught"
...
Exception in thread Thread-3:
Traceback (most recent call last):
File "/usr/lib64/python2.6/threading.py", line 532, in __bootstrap_inner
self.run()
File "/usr/lib64/python2.6/threading.py", line 736, in run
self.function(*self.args, **self.kwargs)
File "<stdin>", line 5, in _timeoutHandler
Exception: Timeout!
那么,我怎样才能在主线程中捕获异常?我应该放弃上下文管理器的想法吗?
1 回答
有一个名为stopit的模块,它使用线程或信号提供可中断的上下文管理器 . 但是,每种方法都有其自身的局限性 . 例如,使用线程,您可以安全地在多线程应用程序中使用 .
看起来它利用C-API函数PyThreadState_SetAsyncExc在所需的线程中异步引发异常 .
这是一个示例用法(取自他们的文档):