最近在项目中调用第三方接口时候,常常会出现请求超时的状况,或者参数的问题致使调用异代码异常。针对超时异常,查询了python 相关文档,没有并发现完善的包来根据用户自定义
的时间来抛出超时异常的模块。因此本身干脆本身来实现一个自定义的超时异常。目前找到了两种方式来实现超时异常的功能(signal.alarm()、threading实现超时异常)
方法1 thread + time
原理:将要调用的功能函数放入子线程,经过设定子线程的阻塞时间,超时则主线程并不会等待子线程的执行。主线程退出,子线程就不存在了。
核心就是在程序中添加 join()方法,用于等待线程结束。join()的做用是,在子线程完成运行以前,这个子线程的父线程将会被一直阻塞.
1 # coding=utf-8 2 import threading 3 import time 4 5 6 def myFunc(): 7 time.sleep(4) 8 print("myFunc执行了") 9 10 11 if __name__ == '__main__': 12 t = threading.Thread(target=myFunc) 13 t.setDaemon(True) 14 t.start() 15 16 t.join(2) 17 print("it's over")
执行结果:
it's overpython
能够看出,当主线程执行到2秒时候,结束退出。子线程尚未结束,没有执行完及被强制退出linux
1 # coding=utf-8 2 import threading 3 import time 4 5 6 def myFunc(): 7 time.sleep(1) 8 print("myFunc执行了") 9 10 11 if __name__ == '__main__': 12 t = threading.Thread(target=myFunc) 13 t.setDaemon(True) 14 t.start() 15 16 t.join(2) 17 print("it's over")
显示结果:
myFunc执行了
it's over并发
能够看出,子线程结束时,用时1秒,没有超过主线程设定的3秒,因此主线程与子线程都被执行了函数
方法 2 signal.alarm() ,注意两点:一是signal信号机制要在linux上才能运行; 二是signal信号在主线程中才会会起做用spa
1 import signal 2 import time 3 4 5 # Define signal handler function 6 def myHandler(signum, frame): 7 exit("TimeoutError") 8 9 10 def test_fun(): 11 # time.sleep(3) 12 int("afsdf") 13 a = 2 + 3 14 15 return a 16 17 18 19 if __name__ == '__main__': 20 try: 21 signal.signal(signal.SIGALRM, myHandler) 22 signal.alarm(2) 23 test = test_fun() 24 print(test) 25 signal.alarm(0) 26 except Exception as ret: 27 print("msg:", ret)
执行结果:
当 time.sleep(3) 时,会抛出TimeoutError的异常
当 test_fun 里面出现 int("afsdf")时, 会抛出 ValueError("invalid literal for int() with base 10: 'afsdf'",))
当test_fun函数执行的时间小于2 秒时,就会返回函数对应的值线程
方法3 带有返回值的超时异常,能够经过建立thread类的方式来进行捕捉code
1 import threading 2 import sys 3 import time 4 5 6 class Dispacher(threading.Thread): 7 def __init__(self, fun, args): 8 threading.Thread.__init__(self) 9 self.setDaemon(True) 10 self.result = None 11 self.error = None 12 self.fun = fun 13 self.args = args 14 15 self.start() 16 17 def run(self): 18 try: 19 self.result = self.fun(self.args) 20 except: 21 self.error = sys.exc_info() 22 23 24 def test_fun(i): 25 # time.sleep(4) 26 a = i*i 27 # b 29 return a 30 def main_fun(): 31 c = Dispacher(test_fun, 2) 32 c.join(2) 33 34 if c.isAlive(): 35 return "TimeOutError" 36 elif c.error: 37 return c.error[1] 38 t = c.result 39 return t 40 41 if __name__ == '__main__': 42 fun = main_fun() 43 print(fun)
显示结果:
test_fun 执行时间大于设置的2秒时,会抛出TimeOutError test_fun 执行时间小于设置的2秒时,而且函数正常执行时,显示:4 test_fun 里面出现好比 “b” 时,会抛出 global name 'b' is not defined 的异常