超时装饰器
1. 使用信号实现⚓
def setTimeout(second=5):
def wrap(function):
def handle(signal, frame):
raise RuntimeError
def timing(second):
while second != 0:
second -= 1
time.sleep(1)
def to_do(*args, **kwargs):
try:
signal.signal(signal.SIGALRM, handle)
signal.alarm(second)
re = function(*args, **kwargs)
signal.alarm(0)
return re
except RuntimeError as e:
raise e
return to_do
return wrap
缺陷:
- 如果function是当前线程阻塞式的,那么这段代码不会发生作用。
signal
不能在本身就是子线程中使用
2. 使用多线程实现⚓
# author : ignorantshr
# create_date : 2020/04/23 8:51 AM
# description : python3 超时装饰器
import functools
import threading
import time
def timeout(seconds=0):
assert type(seconds) == int
def decorator(func):
re = None
def timer():
count = seconds
while count > 0:
time.sleep(1)
count -= 1
def run(*args, **kwargs):
nonlocal re # 对于 python2,可以使用 global 关键字代替
re = func(*args, **kwargs)
@functools.wraps(func)
def wrapper(*args, **kwargs):
timer_thread = threading.Thread(target=timer)
# quit with main thread
timer_thread.setDaemon(True)
timer_thread.start()
run_thread = threading.Thread(target=run, args=args, kwargs=kwargs)
# quit with main thread
run_thread.setDaemon(True)
run_thread.start()
while True:
if run_thread.is_alive():
# still running
if timer_thread.is_alive():
time.sleep(1) # Query once per second
continue
else:
# run func timeout
return None
# run func done
else:
return re
return wrapper
return decorator
@timeout(5)
def f1():
print('running f1')
def f2():
print('running f2')
if __name__ == '__main__':
"""
这两种方式是一样的
"""
f1()
print('================')
deco = timeout(5)
wrapper_ = deco(f2)
wrapper_()