超时function,如果它需要太长的时间才能完成
我有一个shell脚本,通过一个包含URL的文本文件进行循环,我想要访问并截取它的截图。
所有这一切都很简单。 该脚本初始化一个类,当运行时创build列表中每个站点的屏幕截图。 有些网站需要很长时间才能加载,有些网站可能根本无法加载。 所以我想在一个超时脚本中包装screengrabber函数,使得函数在10秒内不能完成的情况下返回False
。
我满足于最简单的解决scheme,也许设置一个asynchronous计时器,无论实际发生在函数内部,10秒钟后都会返回False。
信号文档中描述了超时操作的过程。
基本的想法是使用信号处理程序在某个时间间隔内设置闹钟,并在该定时器到期后引发exception。
请注意,这只适用于UNIX。
这是一个创build装饰器的实现(将以下代码保存为timeout.py
)。
from functools import wraps import errno import os import signal class TimeoutError(Exception): pass def timeout(seconds=10, error_message=os.strerror(errno.ETIME)): def decorator(func): def _handle_timeout(signum, frame): raise TimeoutError(error_message) def wrapper(*args, **kwargs): signal.signal(signal.SIGALRM, _handle_timeout) signal.alarm(seconds) try: result = func(*args, **kwargs) finally: signal.alarm(0) return result return wraps(func)(wrapper) return decorator
这将创build一个名为@timeout
的装饰器,可以应用于任何长时间运行的函数。
所以,在你的应用程序代码中,你可以像这样使用装饰器:
from timeout import timeout # Timeout a long running function with the default expiry of 10 seconds. @timeout def long_running_function1(): ... # Timeout after 5 seconds @timeout(5) def long_running_function2(): ... # Timeout after 30 seconds, with the error "Connection timed out" @timeout(30, os.strerror(errno.ETIMEDOUT)) def long_running_function3(): ...
我使用with
语句重写了David的答案,它允许你这样做:
with timeout(seconds=3): time.sleep(4)
这会引发TimeoutError。
代码仍然使用signal
,因此只有UNIX:
import signal class timeout: def __init__(self, seconds=1, error_message='Timeout'): self.seconds = seconds self.error_message = error_message def handle_timeout(self, signum, frame): raise TimeoutError(self.error_message) def __enter__(self): signal.signal(signal.SIGALRM, self.handle_timeout) signal.alarm(self.seconds) def __exit__(self, type, value, traceback): signal.alarm(0)