Python控制函数运行时间
在线程中主动判断时间的运行长度
import requests, datetime, time
import threading
class MyThread(threading.Thread):
def __init__(self, target, args=()):
"""
why: 因为threading类没有返回值,因此在此处重新定义MyThread类,使线程拥有返回值
此方法来源 https://www.cnblogs.com/hujq1029/p/7219163.html?utm_source=itdadao&utm_medium=referral
"""
super(MyThread, self).__init__()
self.func = target
self.args = args
def run(self):
# 接受返回值
self.result = self.func(*self.args)
def get_result(self):
# 线程不结束,返回值为None
try:
return self.result
except Exception:
return None
# 为了限制真实请求时间或函数执行时间的装饰器
def limit_decor(limit_time):
"""
:param limit_time: 设置最大允许执行时长,单位:秒
:return: 未超时返回被装饰函数返回值,超时则返回 None
"""
def functions(func):
# 执行操作
def run(*params):
thre_func = MyThread(target=func, args=params)
# 主线程结束(超出时长),则线程方法结束
thre_func.setDaemon(True)
thre_func.start()
# 计算分段沉睡次数
sleep_num = int(limit_time // 1)
sleep_nums = round(limit_time % 1, 1)
# 多次短暂沉睡并尝试获取返回值
for i in range(sleep_num):
time.sleep(1)
infor = thre_func.get_result()
if infor:
return infor
time.sleep(sleep_nums)
# 最终返回值(不论线程是否已结束)
if thre_func.get_result():
return thre_func.get_result()
else:
return"请求超时" #超时返回 可以自定义
return run
return functions
#接口函数
def a1():
print("开始请求接口")
#这里把逻辑封装成一个函数,使用线程调用
a_theadiing = MyThread(target=a2)
a_theadiing.start()
a_theadiing.join()
#返回结果
a = a_theadiing.get_result()
print("请求完成")
return a
@limit_decor(3) #超时设置为3s 2s逻辑未执行完毕返回接口超时
def a2():
print("开始执行")
time.sleep(2)
print("执行完成")
a=2
return a
# 程序入口 未超时返回a的值 超时返回请求超时
if __name__ == '__main__':
a = a1() #调用接口(这里把函数a1看做一个接口)
print(a)
使用信号模块signal (推荐使用,略复杂但好用)
def set_timeout(num):
def wrap(func):
def handle(signum, frame): # 收到信号 SIGALRM 后的回调函数,第一个参数是信号的数字,第二个参数是the interrupted stack frame.
raise RuntimeError
def to_do(*args):
try:
signal.signal(signal.SIGALRM, handle) # 设置信号和回调函数
signal.alarm(num) # 设置 num 秒的闹钟
print('start alarm signal.')
r = func(*args)
print('close alarm signal.')
signal.alarm(0) # 关闭闹钟
return r
except RuntimeError as e:
return "超时啦"
return to_do
return wrap
@set_timeout(2) # 限时 2 秒超时
def connect(): # 要执行的函数
time.sleep(3) # 函数执行时间,写大于2的值,可测试超时
return "完成"
if __name__ == '__main__':
a = connect()
参考文档