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

yipeiwu_com6年前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实现的文本编辑器功能示例

Python实现的文本编辑器功能示例

本文实例讲述了Python实现的文本编辑器功能。分享给大家供大家参考,具体如下: wxpython实现的文本编辑器 效果如下: 主要功能: 1.编辑保存文本,打开修改文本 2.常用快捷...

基于数据归一化以及Python实现方式

数据归一化: 数据的标准化是将数据按比例缩放,使之落入一个小的特定区间,去除数据的单位限制,将其转化为无量纲的纯数值,便于不同单位或量级的指标能够进行比较和加权。 为什么要做归一化: 1...

python实现画出e指数函数的图像

python实现画出e指数函数的图像

这里用Python逼近函数y = exp(x);同样使用泰勒函数去逼近: exp(x) = 1 + x + (x)^2/(2!) + .. + (x)^n/(n!) + ... #!...

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

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

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

pandas 数据实现行间计算的方法

如下所示: ###方法1:用shift函数,不用通过循环 import pandas as pd import numpy as np import matplotlib as p...