- eventlet (简单好用，但要修改函数)
import requests
import time
import eventlet
eventlet.monkey_patch() # 必须加这条代码
try:
# Dontla 20200421 超时将抛出异常
with eventlet.Timeout(2, True): # 设置超时时间为2秒
time.sleep(4)
r = requests.get("http://10.101.35.249/", verify=False)
print('error')
print('程序运行未超时！')
except eventlet.timeout.Timeout:
# except: # （或，两种都行，注意不能用except Exception，因为Exception是异常基类，我们这个超时异常未包含在它里面）
print('程序运行超时！')
- 装饰器（自己写），不用修改原来的函数
from threading import Thread
import time
import inspect
import ctypes
import sys
sys.setrecursionlimit(1000000) #不加不知道为啥出现递归错误
class TimeoutException(Exception):
pass
def _async_raise(tid, exctype):
"""raises the exception, performs cleanup if needed"""
tid = ctypes.c_long(tid)
if not inspect.isclass(exctype):
exctype = type(exctype)
res = ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, ctypes.py_object(exctype))
if res == 0:
raise ValueError("invalid thread id")
elif res != 1:
# """if it returns a number greater than one, you're in trouble,
# and you should call it again with exc=NULL to revert the effect"""
ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, None)
raise SystemError("PyThreadState_SetAsyncExc failed")
def stop_thread(thread):
_async_raise(thread.ident, SystemExit)
def timelimited(timeout):
def decorator(function):
def decorator2(*args, **kwargs):
class TimeLimited(Thread):
def __init__(self, _error=None, ):
Thread.__init__(self)
self._error = _error
def run(self):
try:
self.result = function(*args, **kwargs)
except Exception as e:
self._error = str(e)
def _stop(self):
if self.is_alive():
stop_thread(self)
t = TimeLimited()
t.start()
t.join(timeout)
if isinstance(t._error, TimeoutException):
t._stop()
raise TimeoutException('timeout for %s' % (repr(function)))
if t.is_alive():
t._stop()
raise TimeoutException('timeout for %s' % (repr(function)))
if t._error is None:
return t.result
return decorator2
return decorator
@timelimited(5) # 设置运行超时时间2S
def fn_1(secs):
time.sleep(secs)
print('fn_1')
return 'Finished without timeout'
def do_something_after_timeout():
print('Time out!')
if __name__ == "__main__":
try:
print(fn_1(3)) # 设置函数执行3S
except TimeoutException as e:
print(str(e))
do_something_after_timeout()
- 装饰器（开源），不用修改原来的函数
# pip install func_timeout
https://zhuanlan.zhihu.com/p/39743129
import time
from func_timeout import func_set_timeout
from func_timeout import func_timeout, FunctionTimedOut
@func_set_timeout(2)
def task():
while True:
print('hello world')
time.sleep(1)
if __name__ == '__main__':
try:
task()
except FunctionTimedOut:
print("doit('arg1', 'arg2') could not complete within 5 seconds and was terminated.
")
except Exception as e:
pass
- 装饰器(os._exit(1))，不用修改原来的函数,os._exit(1)这里不好
from functools import wraps
import time
import os
from threading import Thread
def set_time_limit(t):
def auto_quit(t1):
'''此为控制进程超时退出的线程函数'''
time.sleep(t1)
print("time out {}".format(t1))
os._exit(1) #此函数专门用于线程控制主进程退出，有兴趣的可以看一下和sys.exit()的区别
def decorator(f):
'''此函数用于传入被装饰函数f'''
@wraps(f)
def wrapper(*args,**kwargs):
'''装饰器内部遵循的逻辑是：
1.auto_quit先执行完，进程结束
2.被修饰函数f先执行完，auto_quit函数停止执行
3.被修饰函数执行完，下面的代码才能运行
'''
t1=Thread(target=auto_quit,args=(t,)) #此处的t是set_time_limit函数的形参，是auto_quit函数的实参
t2=Thread(target=f,args=args,kwargs=kwargs)
t1.setDaemon(True) #满足第2点
t1.start()
t2.start()
t2.join() #满足第3点
return wrapper
return decorator
@set_time_limit(3)
def a(t):
time.sleep(t)
print("execuate finished")
if __name__=="__main__":
a(1)
a(5)
a(1) # 不会打印了
print("执行完毕")