快捷導航

python實現(xiàn)超時退出的三種方式總結(jié)

更新時間：2022年11月28日 09:41:29 作者：Noah1995

這篇文章主要介紹了python實現(xiàn)超時退出的三種方式總結(jié)，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教

基于signal模塊實現(xiàn)

signal包負責在Python程序內(nèi)部處理信號，典型的操作包括預設(shè)信號處理函數(shù)，暫停并等待信號，以及定時發(fā)出SIGALRM等。

要注意，signal包主要是針對UNIX平臺(比如Linux, MAC OS)，而Windows內(nèi)核中由于對信號機制的支持不充分，所以在Windows上的Python不能發(fā)揮信號系統(tǒng)的功能。

# coding:utf8
import time
import signal
?
?
# 自定義超時異常
class TimeoutError(Exception):
? ? def __init__(self, msg):
? ? ? ? super(TimeoutError, self).__init__()
? ? ? ? self.msg = msg
?
?
def time_out(interval, callback):
? ? def decorator(func):
? ? ? ? def handler(signum, frame):
? ? ? ? ? ? raise TimeoutError("run func timeout")
?
? ? ? ? def wrapper(*args, **kwargs):
? ? ? ? ? ? try:
? ? ? ? ? ? ? ? signal.signal(signal.SIGALRM, handler)
? ? ? ? ? ? ? ? signal.alarm(interval) ? ? ? # interval秒后向進程發(fā)送SIGALRM信號
? ? ? ? ? ? ? ? result = func(*args, **kwargs)
? ? ? ? ? ? ? ? signal.alarm(0) ? ? ? ? ? ? ?# 函數(shù)在規(guī)定時間執(zhí)行完后關(guān)閉alarm鬧鐘
? ? ? ? ? ? ? ? return result
? ? ? ? ? ? except TimeoutError, e:
? ? ? ? ? ? ? ? callback(e)
? ? ? ? return wrapper
? ? return decorator
?
?
def timeout_callback(e):
? ? print(e.msg)
?
?
@time_out(2, timeout_callback)
def task1():
? ? print("task1 start")
? ? time.sleep(3)
? ? print("task1 end")
?
?
@time_out(2, timeout_callback)
def task2():
? ? print("task2 start")
? ? time.sleep(1)
? ? print("task2 end")
?
?
if __name__ == "__main__":
? ? task1()
? ? task2()

輸出：

task1 start
run func timeout
task2 start
task2 end

基于子線程阻塞實現(xiàn)超時

# coding:utf8
import time
import threading
?
?
def callback_func():
? ? print('超時回調(diào)')
?
?
def time_out(interval, callback=None):
? ? def decorator(func):
? ? ? ? def wrapper(*args, **kwargs):
? ? ? ? ? ? t =threading.Thread(target=func, args=args, kwargs=kwargs)
? ? ? ? ? ? t.setDaemon(True) ?# 設(shè)置主線程技術(shù)子線程立刻結(jié)束
? ? ? ? ? ? t.start()
? ? ? ? ? ? t.join(interval) ?# 主線程阻塞等待interval秒
? ? ? ? ? ? if t.is_alive() and callback:
? ? ? ? ? ? ? ? return threading.Timer(0, callback).start() ?# 立即執(zhí)行回調(diào)函數(shù)
? ? ? ? ? ? else:
? ? ? ? ? ? ? ? return
? ? ? ? return wrapper
? ? return decorator
?
?
@time_out(2, callback_func)
def task3(hh):
? ? print('**********task3****************')
? ? for i in range(3):
? ? ? ? time.sleep(1)
? ? ? ? print(i)
? ? ? ? print(hh)
?
?
@time_out(2, callback_func)
def task4(hh):
? ? print('**********task4****************')
? ? for i in range(3):
? ? ? ? # time.sleep(1)
? ? ? ? print(i)
? ? ? ? print(hh)
?
?
if __name__ == '__main__':
? ? task3('參數(shù)')
? ? task4('參數(shù)')

輸出：

**********task3****************
0
參數(shù)
1
參數(shù)
超時回調(diào)
**********task4****************
0
參數(shù)
1
參數(shù)
2
參數(shù)

基于協(xié)程實現(xiàn)

def callback_func():
? ? print('callback')
?
?
def time_out(interval, callback=None):
? ? def decorator(func):
? ? ? ? def wrapper(*args, **kwargs):
? ? ? ? ? ? ########## 該部分必選在requests之前導入
? ? ? ? ? ? import gevent
? ? ? ? ? ? from gevent import monkey
? ? ? ? ? ? monkey.patch_all()
? ? ? ? ? ? ##########
? ? ? ? ? ??
? ? ? ? ? ? try:
? ? ? ? ? ? ? ? gevent.with_timeout(interval, func, *args, **kwargs)
? ? ? ? ? ? except gevent.timeout.Timeout as e:
? ? ? ? ? ? ? ? callback() if callback else None
?
? ? ? ? return wrapper
?
? ? return decorator
?
?
@time_out(3, callback_func)
def func(a, b):
? ? import time
? ? time.sleep(2)
? ? print(a,b)
?
?
func(1, 2)

以上為個人經(jīng)驗，希望能給大家一個參考，也希望大家多多支持腳本之家。

您可能感興趣的文章: