目前可能首选的选项之一就是使用Pythons
多处理(特别是它的 的 proc.join(timeoutTime) 强> 方法 )
模块(见 教程 )
只需复制/粘贴下面的代码示例并运行它。你是追求的吗?
def beBusyFor(noOfSeconds): import time print(" beBusyFor() message: going to rest for", noOfSeconds, "seconds") time.sleep(noOfSeconds) print(" beBusyFor() message: was resting", noOfSeconds, "seconds, now AWAKE") import multiprocessing noOfSecondsBusy = 5; timeoutTime = 3 print("--- noOfSecondsBusy = 5; timeoutTime = 3 ---") proc = multiprocessing.Process(target=beBusyFor, args=(noOfSecondsBusy, )) print("Start beBusyFor()") proc.start() print("beBusyFor() is running") proc.join(timeoutTime) if proc.is_alive(): print(timeoutTime, "seconds passed, beBusyFor() still running, terminate()" ) proc.terminate() else: print("OK, beBusyFor() has finished its work in time.") #:if print() noOfSecondsBusy = 2; timeoutTime = 3 print("--- noOfSecondsBusy = 2; timeoutTime = 3 ---") proc = multiprocessing.Process(target=beBusyFor, args=(noOfSecondsBusy, )) print("Start beBusyFor()") proc.start() print("beBusyFor() started") proc.join(timeoutTime) if proc.is_alive(): print(timeoutTime, "seconds passed, beBusyFor() still running, terminate()" ) proc.terminate() else: print("OK, beBusyFor() has finished its work in time.") #:if
它输出:
--- noOfSecondsBusy = 5; timeoutTime = 3 --- Start beBusyFor() beBusyFor() is running beBusyFor() message: going to rest for 5 seconds 3 seconds passed, beBusyFor() still running, terminate() --- noOfSecondsBusy = 2; timeoutTime = 3 --- Start beBusyFor() beBusyFor() started beBusyFor() message: going to rest for 2 seconds beBusyFor() message: was resting 2 seconds, now AWAKE OK, beBusyFor() has finished its work in time.
另一个我知道的选择是使用a
的 装饰功能 强> 和 的 信号模块 强>
结帐 具有代码来源的网页 我在这里提供(只需要一个小调整就可以在Python 3.6上运行):
import signal class TimeoutError(Exception): def __init__(self, value = "Timed Out"): self.value = value def __str__(self): return repr(self.value) def timeout(seconds_before_timeout): def decorate(f): def handler(signum, frame): raise TimeoutError() def new_f(*args, **kwargs): old = signal.signal(signal.SIGALRM, handler) signal.alarm(seconds_before_timeout) try: result = f(*args, **kwargs) finally: signal.signal(signal.SIGALRM, old) signal.alarm(0) return result # new_f.func_name = f.func_name new_f.__name__ = f.__name__ return new_f return decorate # Try it out: import time @timeout(5) def mytest(): print( "mytest() message: Started" ) for i in range(1,10): time.sleep(1) print( "mytest() message: %d seconds have passed" % i ) try: mytest() except TimeoutError as e: print("stopped executing mytest() because it", e) print("continuing script execution past call of mytest()")
上面的代码输出:
mytest() message: Started mytest() message: 1 seconds have passed mytest() message: 2 seconds have passed mytest() message: 3 seconds have passed mytest() message: 4 seconds have passed stopped executing mytest() because it 'Timed Out' continuing script execution past call of mytest()