详解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设计】。

相关文章

Mac在python3环境下安装virtualwrapper遇到的问题及解决方法

前言 我在使用mac安装virtualwrapper的时候遇到了问题,搞了好长时间,才弄好,在这里总结一下分享出来,供遇到相同的问题的朋友使用,少走些弯路。 问题说明: Mac默认系...

python访问系统环境变量的方法

本文实例讲述了python访问系统环境变量的方法。分享给大家供大家参考。具体如下: #-------------------------------- # Name: en...

python修改txt文件中的某一项方法

python修改txt文件中的某一项方法

在做task中,需要将TXT文本中的某一项注释修改,但是python对txt文本只有写入和读取两种操作。 我采用的方法是: 1.读取txt文件,将每一行数据,加入新建立的list中。 2...

Python使用requests提交HTTP表单的方法

Python的requests库, 其口号是HTTP for humans,堪称最好用的HTTP库。 使用requests库,可以使用数行代码实现自动化的http操作。以http pos...

pytorch中torch.max和Tensor.view函数用法详解

torch.max() 1. torch.max()简单来说是返回一个tensor中的最大值。 例如: >>> si=torch.randn(4,5) >&g...