python程序超時處理方式

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

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

python程序超時處理

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

其實(shí)就是加了一個裝飾器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)行時間超過6s 就報錯。

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

如果我們設(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超時處理方法介紹，Eventlet 和 func-timeout

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

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

以上場景中，如果我們在程序中設(shè)置了超時處理，那將會為我們的程序節(jié)約很多運(yùn)行時間。

1.eventlet

安裝：

pip install eventlet

eventlet在代碼中的使用：

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

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

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è)置了超時時間3
        print('父進(jìn)程：{}'.format(os.getpid()))
        p = Process(target=my_test, args=('test', ))        # 調(diào)用了子進(jìn)程，子進(jìn)程內(nèi)有等待10秒的步驟，等待完才會打印'子進(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è)置的超時時間3秒沒有生效。這種情況，我們只能將超時處理步驟加到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('等待時間結(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ù)的超時處理方式
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('子程序超時')
# 執(zhí)行以上程序輸出：
# 父進(jìn)程：29908
# 結(jié)束
# 子進(jìn)程運(yùn)行中，name=test，pid=40280
# 子進(jìn)程已經(jīng)結(jié)束
# 說明：my_test函數(shù)執(zhí)行需要4s，設(shè)置的超時時間大于4s，因此子進(jìn)程正常執(zhí)行；若設(shè)置my_test函數(shù)執(zhí)行時間大于5s，則報錯，代碼中嘗試了通過捕獲異常來讓程序正常運(yùn)行退出，但超時異常時發(fā)生在子進(jìn)程中的，因此捕獲異常失敗，此時的報錯見圖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)行超時')
# 輸出：
# 進(jìn)程運(yùn)行中，name=test，pid=3364
# Function my_test (args=('test',)) (kwargs={}) timed out # after 3.000000 seconds.
# 進(jìn)程運(yùn)行超時
# 說明：func_timeout()第一個參數(shù)為超時時間，第二個參數(shù)是調(diào)用的函數(shù)，注意這里只調(diào)用函數(shù)名，沒有(), 第三個參數(shù)是調(diào)用函數(shù)的參數(shù)。這里使用的異常捕獲也不能解決2.1中的報錯問題。