详解python中自定义超时异常的几种方法

Z时代
2024-01-10
分类：综合

最近在项目中调用第三方接口时候，经常会出现请求超时的情况，或者参数的问题导致调用异代码异常。针对超时异常，查询了python 相关文档，没有并发现完善的包来根据用户自定义的时间来抛出超时异常的模块。所以自己干脆自己来实现一个自定义的超时异常。目前找到了两种方式来实现超时异常的功能（signal.alarm()、threading实现超时异常）

方法1 thread + time　

原理：将要调用的功能函数放入子线程，通过设定子线程的阻塞时间，超时则主线程并不会等待子线程的执行。主线程退出，子线程就不存在了。

核心就是在程序中添加 join()方法，用于等待线程结束。join()的作用是，在子线程完成运行之前，这个子线程的父线程将会被一直阻塞.


# coding=utf-8
import threading
import time
def myFunc():
  time.sleep(4)
  print("myFunc执行了")
if __name__ == '__main__':
  t = threading.Thread(target=myFunc)
  t.setDaemon(True)
  t.start()
  t.join(2)
  print("it's over")

执行结果：

it's over

可以看出，当主线程执行到2秒时候，结束退出。子线程还没有结束，没有执行完及被强制退出


# coding=utf-8
import threading
import time
def myFunc():
  time.sleep(1)
  print("myFunc执行了")
if __name__ == '__main__':
  t = threading.Thread(target=myFunc)
  t.setDaemon(True)
  t.start()
  t.join(2)
  print("it's over")

显示结果：

myFunc执行了
it's over

可以看出，子线程结束时，用时1秒，没有超过主线程设定的3秒，所以主线程与子线程都被执行了

方法 2 signal.alarm() ，注意两点：一是signal信号机制要在linux上才能运行；二是signal信号在主线程中才会会起作用


import signal
import time
# Define signal handler function
def myHandler(signum, frame):
  exit("TimeoutError")
def test_fun():
  # time.sleep(3)
  int("afsdf")
  a = 2 + 3
  return a
if __name__ == '__main__':
  try:
    signal.signal(signal.SIGALRM, myHandler)
    signal.alarm(2)
    test = test_fun()
    print(test)
    signal.alarm(0)
  except Exception as ret:
    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类的方式来进行捕捉


import threading
import sys
import time
class Dispacher(threading.Thread):
  def __init__(self, fun, args):
    threading.Thread.__init__(self)
    self.setDaemon(True)
    self.result = None
    self.error = None
    self.fun = fun
    self.args = args
    self.start()
  def run(self):
    try:
      self.result = self.fun(self.args)
    except:
      self.error = sys.exc_info()
def test_fun(i):
  # time.sleep(4)
  a = i*i
  # b 　　　
　　return a
def main_fun():
  c = Dispacher(test_fun, 2)
  c.join(2)
  if c.isAlive():
    return "TimeOutError"
  elif c.error:
    return c.error[1]
  t = c.result
  return t
if __name__ == '__main__':
  fun = main_fun()
  print(fun)