• 欢迎访问搞代码网站,推荐使用最新版火狐浏览器和Chrome浏览器访问本网站!
  • 如果您觉得本站非常有看点,那么赶紧使用Ctrl+D 收藏搞代码吧

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

python 搞代码 4年前 (2022-01-09) 51次浏览 已收录 0个评论

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

方法1 thread + time 

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

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

# coding=utf-8
import threading
import time


def myFunc():
  time.sleep(4)
  print("myFunc执行了")


if __n<strong style="color:transparent">本文来源gao@daima#com搞(%代@#码@网&</strong>ame__ == '__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)

显示结果:

test_fun 执行时间大于设置的2秒时,会抛出TimeOutError
test_fun 执行时间小于设置的2秒时,并且函数正常执行时,显示:4
test_fun 里面出现比如 “b” 时,会抛出 global name ‘b’ is not defined 的异常

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持搞代码


搞代码网(gaodaima.com)提供的所有资源部分来自互联网,如果有侵犯您的版权或其他权益,请说明详细缘由并提供版权或权益证明然后发送到邮箱[email protected],我们会在看到邮件的第一时间内为您处理,或直接联系QQ:872152909。本网站采用BY-NC-SA协议进行授权
转载请注明原文链接:详解python中自定义超时异常的几种方法
喜欢 (0)
[搞代码]
分享 (0)
发表我的评论
取消评论

表情 贴图 加粗 删除线 居中 斜体 签到

Hi,您需要填写昵称和邮箱!

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址