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设计】的支持。如果你想了解更多相关内容请查看下面相关链接

相关文章

wxPython窗口中文乱码解决方法

本文实例讲述了wxPython窗口中文乱码解决方法,分享给大家供大家参考。具体方法如下: 文件保存为 utf-8 文件开头添加 # -*- coding: utf-8 -*- 在有中文字...

Python中字符编码简介、方法及使用建议

1. 字符编码简介 1.1. ASCII ASCII(American Standard Code for Information Interchange),是一种单字节的编码。计算机世...

python处理文本文件并生成指定格式的文件

import os import sys import string #以指定模式打开指定文件,获取文件句柄 def getFileIns(filePath,model)...

将python依赖包打包成window下可执行文件bat方式

1、 打开一个记事本,将需要安装的第三方python依赖包写入文件,比如:需要安装urllib3、flask、bs4三个python库(替换成你想要安装的库,每个库之间用空格隔开),输入...

CentOS中升级Python版本的方法详解

CentOS升级Python2.6到Pythno2.7 最近在Linode上弄Python、出现ValueError: zero length field name in format这...