python程序超時(shí)處理方式

更新時(shí)間：2023年08月17日 15:10:02 作者：何小義的AI進(jìn)階路

這篇文章主要介紹了python程序超時(shí)處理方式,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教

python程序超時(shí)處理

下面這個(gè)小案例，實(shí)現(xiàn)對(duì)某個(gè)函數(shù)超時(shí)處理：

其實(shí)就是加了一個(gè)裝飾器timeout

import time
import timeout_decorator
@timeout_decorator.timeout(6)
def justtest():
    time.sleep(5)
    return 5
aaa = justtest()
print aaa

上面代碼我們規(guī)定若函數(shù)運(yùn)行時(shí)間超過6s 就報(bào)錯(cuò)。

5s是可以的，所以正常運(yùn)行，運(yùn)行結(jié)果：

如果我們?cè)O(shè)定程序不可超過3s，

import time
import timeout_decorator
@timeout_decorator.timeout(3)
def justtest():
    time.sleep(5)
    return 5
aaa = justtest()
print aaa

運(yùn)行結(jié)果：

python超時(shí)處理方法介紹，Eventlet 和 func-timeout

有一些場(chǎng)景，需要我們控制程序或者腳本的運(yùn)行時(shí)間。

自動(dòng)化用例中的某一個(gè)步驟需要在規(guī)定時(shí)間內(nèi)完成才算有效；
使用線程池控制線程運(yùn)行時(shí)，我們不希望每個(gè)線程運(yùn)行時(shí)間超過某一值；
在爬蟲時(shí)發(fā)送的request請(qǐng)求，我們希望在規(guī)定時(shí)間內(nèi)無響應(yīng)則跳過該請(qǐng)求；
當(dāng)寫入了一條錯(cuò)誤的sql查詢語句可能會(huì)導(dǎo)致該查詢語句一直運(yùn)行，導(dǎo)致其他查詢語句阻塞時(shí)等等場(chǎng)景

以上場(chǎng)景中，如果我們?cè)诔绦蛑性O(shè)置了超時(shí)處理，那將會(huì)為我們的程序節(jié)約很多運(yùn)行時(shí)間。

1.eventlet

安裝：

pip install eventlet

eventlet在代碼中的使用：

import time
import eventlet                     # 導(dǎo)入eventlet模塊
eventlet.monkey_patch()             # 引入patch
with eventlet.Timeout(3, False):    # 設(shè)置超時(shí)時(shí)間3秒
    time.sleep(4)                   # 等待4秒
    print('結(jié)果1')                   # 因?yàn)榈却?秒，大于超時(shí)時(shí)間，所以'結(jié)果1‘不打印
print('結(jié)果2')
# 運(yùn)行以上程序輸出：
# 結(jié)果2
with eventlet.Timeout(3, False):    # 設(shè)置超時(shí)時(shí)間3秒
    time.sleep(2)                   # 等待2秒
    print('結(jié)果3')                   # 因?yàn)榈却?秒，小于超時(shí)時(shí)間，所以'結(jié)果3‘也會(huì)打印
print('結(jié)果4')
# 運(yùn)行以上程序輸出：
# 結(jié)果3
# 結(jié)果4

注意：eventlet方法中只要進(jìn)行了子進(jìn)程的調(diào)用，超時(shí)設(shè)置就會(huì)失效。示例：

import time
import os
import eventlet                     # 導(dǎo)入eventlet模塊
from multiprocessing import Process
def my_test(name):
    print('子進(jìn)程運(yùn)行中，name={}，pid={}'.format(name, os.getpid()))
    time.sleep(10)
    print('子進(jìn)程已經(jīng)結(jié)束')
if __name__ == '__main__':
    eventlet.monkey_patch()                                 # 引入patch
    with eventlet.Timeout(3, False):                        # 這里設(shè)置了超時(shí)時(shí)間3
        print('父進(jìn)程：{}'.format(os.getpid()))
        p = Process(target=my_test, args=('test', ))        # 調(diào)用了子進(jìn)程，子進(jìn)程內(nèi)有等待10秒的步驟，等待完才會(huì)打印'子進(jìn)程已經(jīng)結(jié)束‘
        print('子進(jìn)程準(zhǔn)備執(zhí)行---')
        p.start()
# 執(zhí)行以上程序輸出：
# 父進(jìn)程：11548
# 子進(jìn)程準(zhǔn)備執(zhí)行 - --
# 子進(jìn)程運(yùn)行中，name = test，pid = 1996
# 子進(jìn)程已經(jīng)結(jié)束

從以上示例可以看出，設(shè)置的超時(shí)時(shí)間3秒沒有生效。這種情況，我們只能將超時(shí)處理步驟加到my_test函數(shù)中即可：

def my_test(name):
    print('子進(jìn)程運(yùn)行中，name={}，pid={}'.format(name, os.getpid()))
    eventlet.monkey_patch()                                 # 引入patch
    with eventlet.Timeout(3, False):
        time.sleep(10)
        print('等待時(shí)間結(jié)束')
    print('子進(jìn)程已經(jīng)結(jié)束')

2.func-timeout

安裝：

pip install func-timeout

func-timeout 在代碼中的使用，func-timeout主要有兩中方法：func_timeout，func_set_timeout

2.1 func_set_timeout 用法

# func_set_timeout作為裝飾器使用，來作為目標(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í)')
# 執(zhí)行以上程序輸出：
# 父進(jìn)程：29908
# 結(jié)束
# 子進(jìn)程運(yùn)行中，name=test，pid=40280
# 子進(jìn)程已經(jīng)結(jié)束
# 說明：my_test函數(shù)執(zhí)行需要4s，設(shè)置的超時(shí)時(shí)間大于4s，因此子進(jìn)程正常執(zhí)行；若設(shè)置my_test函數(shù)執(zhí)行時(shí)間大于5s，則報(bào)錯(cuò)，代碼中嘗試了通過捕獲異常來讓程序正常運(yùn)行退出，但超時(shí)異常時(shí)發(fā)生在子進(jìn)程中的，因此捕獲異常失敗，此時(shí)的報(bào)錯(cuò)見圖1。

圖1：

2.2 func_timeout 用法

import time
import os
from func_timeout import func_timeout
from func_timeout import FunctionTimedOut
def my_test(name):
    print('進(jìn)程運(yùn)行中，name={}，pid={}'.format(name, os.getpid()))
    time.sleep(6)
    print('進(jìn)程已經(jīng)結(jié)束')
if __name__ == '__main__':
    try:
        func_timeout(3, my_test, args=('test', ))
    except FunctionTimedOut as e:
        print(e)
        print('進(jìn)程運(yùn)行超時(shí)')
# 輸出：
# 進(jìn)程運(yùn)行中，name=test，pid=3364
# Function my_test (args=('test',)) (kwargs={}) timed out # after 3.000000 seconds.
# 進(jìn)程運(yùn)行超時(shí)
# 說明：func_timeout()第一個(gè)參數(shù)為超時(shí)時(shí)間，第二個(gè)參數(shù)是調(diào)用的函數(shù)，注意這里只調(diào)用函數(shù)名，沒有(), 第三個(gè)參數(shù)是調(diào)用函數(shù)的參數(shù)。這里使用的異常捕獲也不能解決2.1中的報(bào)錯(cuò)問題。