python防止程序超時(shí)的實(shí)現(xiàn)示例
因?yàn)槟硞€(gè)需求,需要在程序運(yùn)行的時(shí)候防止超時(shí)。在網(wǎng)上搜了以下發(fā)現(xiàn)有3種方法
1.@func_set_timeout這個(gè)注解
實(shí)測(cè)可以用,但是用起來(lái)比較麻煩,因?yàn)檫@個(gè)超時(shí)后會(huì)跑出異常,可以直接注解在函數(shù)上但不能注解在單條語(yǔ)句上,我設(shè)想的是把超時(shí)和異常同時(shí)處理,這個(gè)會(huì)拋出新的異常不太符合我的需求
# func_set_timeout作為裝飾器使用,來(lái)作為目標(biāo)函數(shù)的超時(shí)處理方式
import time
import os
from func_timeout import func_set_timeout
@func_set_timeout(5)
def my_test(name):
print('子進(jìn)程運(yùn)行中,name={},pid={}'.format(name, os.getpid()))
time.sleep(4)
print('子進(jìn)程已經(jīng)結(jié)束')
if __name__ == '__main__':
print('父進(jìn)程:{}'.format(os.getpid()))
try:
p = Process(target=my_test, args=('test', ))
p.start()
except TimeoutError as e:
print('子程序超時(shí)')2.多進(jìn)程/多限制的.join
比如下面的代碼
import time
import threading
def worker():
print('worker start')
time.sleep(10)
print('worker end')
t = threading.Thread(target=worker)
t.start()
# 等待線程結(jié)束,不能超過(guò)5秒
t.join(5)
if t.is_alive():
print('worker overtime')
else:
print('worker finished')實(shí)測(cè)發(fā)現(xiàn)運(yùn)行結(jié)果是這樣,到了5秒提示overtime但是程序繼續(xù)運(yùn)行,然后到了10秒程序運(yùn)行結(jié)束,雖然有超時(shí)的效果但是和我設(shè)想的還是有區(qū)別,我想的是時(shí)間到了你就別繼續(xù)運(yùn)行了
3.使用eventlet.Timeout來(lái)實(shí)現(xiàn)
實(shí)測(cè)下來(lái)這個(gè)是最符合我需求的,時(shí)間到了直接跳出,加一個(gè)timeout的變量就能知道是否超時(shí)了。美中不足就是必須加一條語(yǔ)句,eventlet.monkey_patch(),簡(jiǎn)單搜索了下發(fā)現(xiàn)這個(gè)屬于程序在運(yùn)行時(shí)動(dòng)態(tài)對(duì)已有代碼進(jìn)行修改,而不需要修改原始代碼的一個(gè)熱補(bǔ)丁。
使用了eventlet后,同時(shí)處理超時(shí)和異常的代碼如下
"""
用于處理函數(shù)超時(shí)的情況,同時(shí)記錄程序異常
"""
import eventlet
class PyTimer:
"""
用于在處理異常的同時(shí)防止程序超時(shí)
"""
def __init__(self, max_time=5):
"""
初始化
"""
# 必須有這條代碼
eventlet.monkey_patch()
self.errors = ""
# 最大超時(shí)的秒數(shù)
self.max_time = max_time
def set_max_time(self,new_time):
"""
設(shè)置新的超時(shí)秒數(shù)
"""
self.max_time=new_time
def count_time(self, function, *args):
"""
運(yùn)行函數(shù),記錄異常,同時(shí)超時(shí)跳出
"""
return_result = None
timeout = True
current_error = ""
try:
with eventlet.Timeout(self.max_time, False): # 設(shè)置超時(shí)時(shí)間為5秒
return_result = function(*args)
timeout = False
except Exception as e:
current_error = str(e) + "\n"
finally:
if timeout:
current_error += function.__name__ + "超時(shí)\n"
print(current_error)
self.errors += current_error
return return_result
def get_errors(self):
"""
獲取報(bào)錯(cuò)信息
"""
return self.errors
if __name__ == '__main__':
def long_fun(a, b):
import time
print(a, b)
time.sleep(20)
my_timer = PyTimer()
my_timer.count_time(long_fun, 1, 2)到此這篇關(guān)于python防止程序超時(shí)的實(shí)現(xiàn)示例的文章就介紹到這了,更多相關(guān)python防止程序超時(shí)內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
Python搭建自己IP代理池的方法實(shí)現(xiàn)
本文主要介紹了Python搭建自己IP代理池的方法實(shí)現(xiàn),文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2023-02-02
Python字符串內(nèi)置函數(shù)功能與用法總結(jié)
這篇文章主要介紹了Python字符串內(nèi)置函數(shù)功能與用法,結(jié)合實(shí)例形式總結(jié)分析了Python常見(jiàn)字符串操作函數(shù)的功能、分類(lèi)、使用方法及相關(guān)操作注意事項(xiàng),需要的朋友可以參考下2019-04-04
Python中的Sequence類(lèi)型使用實(shí)例
這篇文章主要為大家介紹了Python中的Sequence類(lèi)型使用實(shí)例,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2023-12-12
Django配置kafka消息隊(duì)列的實(shí)現(xiàn)
本文主要介紹了Django配置kafka消息隊(duì)列的實(shí)現(xiàn),文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2023-05-05

