Python裝飾器限制函數(shù)運(yùn)行時間超時則退出執(zhí)行

更新時間：2019年04月09日 15:35:57 作者：-牧野-

今天小編就為大家分享一篇關(guān)于Python裝飾器限制函數(shù)運(yùn)行時間超時則退出執(zhí)行，小編覺得內(nèi)容挺不錯的，現(xiàn)在分享給大家，具有很好的參考價值，需要的朋友一起跟隨小編來看看吧

實(shí)際項目中會涉及到需要對有些函數(shù)的響應(yīng)時間做一些限制，如果超時就退出函數(shù)的執(zhí)行，停止等待。

可以利用python中的裝飾器實(shí)現(xiàn)對函數(shù)執(zhí)行時間的控制。

python裝飾器簡單來說可以在不改變某個函數(shù)內(nèi)部實(shí)現(xiàn)和原來調(diào)用方式的前提下對該函數(shù)增加一些附件的功能，提供了對該函數(shù)功能的擴(kuò)展。

方法一. 使用 signal

# coding=utf-8
import signal
import time
def set_timeout(num, callback):
  def wrap(func):
    def handle(signum, frame): # 收到信號 SIGALRM 后的回調(diào)函數(shù)，第一個參數(shù)是信號的數(shù)字，第二個參數(shù)是the interrupted stack frame.
      raise RuntimeError
    def to_do(*args, **kwargs):
      try:
        signal.signal(signal.SIGALRM, handle) # 設(shè)置信號和回調(diào)函數(shù)
        signal.alarm(num) # 設(shè)置 num 秒的鬧鐘
        print('start alarm signal.')
        r = func(*args, **kwargs)
        print('close alarm signal.')
        signal.alarm(0) # 關(guān)閉鬧鐘
        return r
      except RuntimeError as e:
        callback()
    return to_do
  return wrap
def after_timeout(): # 超時后的處理函數(shù)
  print("Time out!")
@set_timeout(2, after_timeout) # 限時 2 秒超時
def connect(): # 要執(zhí)行的函數(shù)
  time.sleep(3) # 函數(shù)執(zhí)行時間，寫大于2的值，可測試超時
  print('Finished without timeout.')
if __name__ == '__main__':
  connect()

方法一中使用的signal有所限制，需要在linux系統(tǒng)上，并且需要在主線程中使用。方法二使用線程計時，不受此限制。

方法二. 使用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) # 設(shè)置運(yùn)行超時時間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)) # 設(shè)置函數(shù)執(zhí)行3S
  except TimeoutException as e:
    print(str(e))
    do_something_after_timeout()

總結(jié)

以上就是這篇文章的全部內(nèi)容了，希望本文的內(nèi)容對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，謝謝大家對腳本之家的支持。如果你想了解更多相關(guān)內(nèi)容請查看下面相關(guān)鏈接

您可能感興趣的文章: