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

yipeiwu_com6年前Python基础

最近在项目中调用第三方接口时候,经常会出现请求超时的情况,或者参数的问题导致调用异代码异常。针对超时异常,查询了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)

显示结果:

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

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持【听图阁-专注于Python设计】。

相关文章

django使用LDAP验证的方法示例

django使用LDAP验证的方法示例

1.安装Python-LDAP(python_ldap-2.4.25-cp27-none-win_amd64.whl)pip install python_ldap-2.4.25-cp2...

新年福利来一波之Python轻松集齐五福(demo)

新年福利来一波之Python轻松集齐五福(demo)

新的一年又要到来了,各个大的公司又在这年末为大家送上了新春的祝福,支付宝还是延续了这几年的传统,在年末为大家送上了集五福的活动,为了大家能更快更好的扫出来大大的“福”,今天就带领大家利用...

tensor和numpy的互相转换的实现示例

要对tensor进行操作,需要先启动一个Session,否则,我们无法对一个tensor比如一个tensor常量重新赋值或是做一些判断操作,所以如果将它转化为numpy数组就好处理了。下...

Python列表推导式与生成器表达式用法示例

本文实例讲述了Python列表推导式与生成器表达式用法。分享给大家供大家参考,具体如下: 和列表一样,列表推导式也采用方括号[]表示,并且用到了一个简写版的for循环,第一部分是一个生成...

Python编程实现二叉树及七种遍历方法详解

Python编程实现二叉树及七种遍历方法详解

本文实例讲述了Python实现二叉树及遍历方法。分享给大家供大家参考,具体如下: 介绍: 树是数据结构中非常重要的一种,主要的用途是用来提高查找效率,对于要重复查找的情况效果更佳,如二叉...