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

yipeiwu_com5年前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设计】。

相关文章

对numpy和pandas中数组的合并和拆分详解

合并 numpy中 numpy中可以通过concatenate,指定参数axis=0 或者 axis=1,在纵轴和横轴上合并两个数组。 import numpy as np impo...

基于Python检测动态物体颜色过程解析

基于Python检测动态物体颜色过程解析

本篇文章将通过图片对比的方法检查视频中的动态物体,并将其中会动的物体定位用cv2矩形框圈出来。本次项目可用于树莓派或者单片机追踪做一些思路参考。寻找动态物体也可以用来监控是否有人进入房间...

Python入门之三角函数sin()函数实例详解

描述 sin()返回的x弧度的正弦值。 语法 以下是sin()方法的语法: importmath math.sin(x) 注意:sin()是不能直接访问的,需要导入math模块,...

Python使用matplotlib绘制余弦的散点图示例

Python使用matplotlib绘制余弦的散点图示例

本文实例讲述了Python使用matplotlib绘制余弦的散点图。分享给大家供大家参考,具体如下: 一 代码 import numpy as np import pylab as...

详谈Python基础之内置函数和递归

详谈Python基础之内置函数和递归

一、内置函数 下面简单介绍几个: 1.abs() 求绝对值 2.all() 如果 iterable 的所有元素都为真(或者如果可迭代为空),则返回 True 3.any() 如果 it...