Python装饰器限制函数运行时间超时则退出执行

yipeiwu_com5年前Python基础

实际项目中会涉及到需要对有些函数的响应时间做一些限制,如果超时就退出函数的执行,停止等待。

可以利用python中的装饰器实现对函数执行时间的控制。

python装饰器简单来说可以在不改变某个函数内部实现和原来调用方式的前提下对该函数增加一些附件的功能,提供了对该函数功能的扩展。

方法一. 使用 signal

# coding=utf-8
import signal
import time
def set_timeout(num, callback):
  def wrap(func):
    def handle(signum, frame): # 收到信号 SIGALRM 后的回调函数,第一个参数是信号的数字,第二个参数是the interrupted stack frame.
      raise RuntimeError
    def to_do(*args, **kwargs):
      try:
        signal.signal(signal.SIGALRM, handle) # 设置信号和回调函数
        signal.alarm(num) # 设置 num 秒的闹钟
        print('start alarm signal.')
        r = func(*args, **kwargs)
        print('close alarm signal.')
        signal.alarm(0) # 关闭闹钟
        return r
      except RuntimeError as e:
        callback()
    return to_do
  return wrap
def after_timeout(): # 超时后的处理函数
  print("Time out!")
@set_timeout(2, after_timeout) # 限时 2 秒超时
def connect(): # 要执行的函数
  time.sleep(3) # 函数执行时间,写大于2的值,可测试超时
  print('Finished without timeout.')
if __name__ == '__main__':
  connect()

方法一中使用的signal有所限制,需要在linux系统上,并且需要在主线程中使用。方法二使用线程计时,不受此限制。

方法二. 使用Thread

# -*- coding: utf-8 -*-
from threading import Thread
import time
class TimeoutException(Exception):
  pass
ThreadStop = Thread._Thread__stop
def timelimited(timeout):
  def decorator(function):
    def decorator2(*args,**kwargs):
      class TimeLimited(Thread):
        def __init__(self,_error= None,):
          Thread.__init__(self)
          self._error = _error
        def run(self):
          try:
            self.result = function(*args,**kwargs)
          except Exception,e:
            self._error = str(e)
        def _stop(self):
          if self.isAlive():
            ThreadStop(self)
      t = TimeLimited()
      t.start()
      t.join(timeout)
      if isinstance(t._error,TimeoutException):
        t._stop()
        raise TimeoutException('timeout for %s' % (repr(function)))
      if t.isAlive():
        t._stop()
        raise TimeoutException('timeout for %s' % (repr(function)))
      if t._error is None:
        return t.result
    return decorator2
  return decorator
@timelimited(2) # 设置运行超时时间2S
def fn_1(secs):
  time.sleep(secs)
  return 'Finished without timeout'
def do_something_after_timeout():
  print('Time out!')
if __name__ == "__main__":
  try:
    print(fn_1(3)) # 设置函数执行3S
  except TimeoutException as e:
    print(str(e))
    do_something_after_timeout()

总结

以上就是这篇文章的全部内容了,希望本文的内容对大家的学习或者工作具有一定的参考学习价值,谢谢大家对【听图阁-专注于Python设计】的支持。如果你想了解更多相关内容请查看下面相关链接

相关文章

Python使用base64模块进行二进制数据编码详解

前言 昨天团队的学妹来问关于POP3协议的问题,所以今天稍稍研究了下POP3协议的格式和Python里面的poplib。而POP服务器往回传的数据里有一部分需要用到Base64进行解码,...

Flask 让jsonify返回的json串支持中文显示的方法

Flask 让jsonify返回的json串支持中文显示的方法

用flask时遇到了返回字符串支持中文显示的问题,在web端显示的是utf-8的编码,而不是中文,如下图。 虽然不影响接口的读取,但是可读性太差,于是研究了一下怎么直接显示成中文。最后...

python3.8下载及安装步骤详解

python3.8下载及安装步骤详解

1.操作系统:Windows7 64bit Python版本:3.8下载地址:https://www.python.org/downloads/release/python-380/,选...

Python GUI布局尺寸适配方法

如下所示: #coding=utf-8 #布局自定义尺寸 from tkinter import * class App: def __init__(self,master...

使用python 和 lint 删除项目无用资源的方法

有部分老项目是在Eclipse环境开发的,最近公司要求应用瘦身,老项目也在其中。如果在 AS 下开发就不会有这样的问题,但是在 Eclipse 中就不太方便了,于是就写了这个脚本。第一次...