限制函數(shù)調(diào)用的最大時間是一種非常實用的技術(shù)手段，能夠幫助開發(fā)者更好地控制程序的行為，提升用戶體驗，同時確保系統(tǒng)的穩(wěn)定性和可靠性。

func-timeout

func-timeout 是一個 Python 庫，允許為函數(shù)設(shè)置超時時間，防止代碼長時間運行或無限阻塞。它適用于需要強制限制執(zhí)行時間的場景，例如網(wǎng)絡(luò)請求、計算密集型任務(wù)或可能出現(xiàn)死循環(huán)的代碼。

1. 安裝 func-timeout

可以使用 pip 安裝：

pip install func-timeout

2. 基本用法

最常用的方式是 func_timeout，它允許在指定的時間內(nèi)運行一個函數(shù)，超時則拋出異常。

from func_timeout import func_timeout, FunctionTimedOut
import time

def long_running_task():
    time.sleep(5)  # 模擬長時間運行的任務(wù)
    return "Task completed"
try:
    result = func_timeout(3, long_running_task)  # 設(shè)置3秒超時
    print(result)
except FunctionTimedOut:
    print("Function execution timed out!")

解釋：

• func_timeout(3, long_running_task)：嘗試在 3 秒內(nèi)運行 long_running_task
• FunctionTimedOut 異常表示函數(shù)超時未完成

也可以使用裝飾器方式為函數(shù)設(shè)定超時：

from func_timeout import func_set_timeout
import time

@func_set_timeout(2)  # 限制該函數(shù)的運行時間為2秒
def long_task():
    time.sleep(5)  # 任務(wù)實際需要5秒
    return "Finished"

try:
    print(long_task())
except FunctionTimedOut:
    print("Function execution timed out!")

• 這種方式適用于需要多次調(diào)用的函數(shù)，避免每次調(diào)用都手動設(shè)置超時。
• func-timeout 本質(zhì)上還是依賴 多線程 或 多進(jìn)程 實現(xiàn)超時控制，在某些情況下可能不適用于主線程（如 Jupyter Notebook）。它也不能用于 main 線程內(nèi)的 while True 死循環(huán)，因為 Python 的 GIL 可能會影響信號處理。

自定義進(jìn)程

除了使用上面的庫，也可以自己使用一個進(jìn)程來計時和檢測超時，另一個進(jìn)程來調(diào)用 Python 函數(shù)。

以下是具體實現(xiàn)代碼：

import time
from itertools import count
from multiprocessing import Process

def inc_forever():
    print('Starting function inc_forever()...')
    while True:
        time.sleep(1)
        print(next(counter))

def return_zero():
    print('Starting function return_zero()...')
    return 0

if __name__ == '__main__':
    # counter 是一個無限迭代器
    counter = count(0)

    p1 = Process(target=inc_forever, name='Process_inc_forever')
    p2 = Process(target=return_zero, name='Process_return_zero')
    p1.start()
    p2.start()
    p1.join(timeout=5)
    p2.join(timeout=5)
    p1.terminate()
    p2.terminate()

if p1.exitcode is None:
    print(f'Oops, {p1} timeouts!')

if p2.exitcode == 0:
    print(f'{p2} is luck and finishes in 5 seconds!')

運行結(jié)果如下：

Starting function inc_forever()...
Starting function return_zero()...
0
1
2
3
4
Oops, <Process(Process_inc_forever, started)> timeouts!
<Process(Process_return_zero, stopped)> is luck and finishes in 5 seconds!

從退出碼可以看出，inc_forever() 函數(shù)超時了（退出碼為 None），而 return_zero() 函數(shù)在 5 秒內(nèi)成功完成。

subprocess

參數(shù)設(shè)置超時

從 Python 3.5 開始，subprocess 模塊提供了一個便捷且推薦使用的 run() API，它內(nèi)置了超時支持。

以下是示例代碼：

import subprocess

r = subprocess.run(['echo', 'hello timeout'], timeout=5)
print(
    f'''type(r)={type(r)},
    r.args={r.args},
    r.returncode={r.returncode},
    r.stdout={r.stdout},
    r.stderr={r.stderr}''')
try:
    r = subprocess.run(['ping', 'www.google.com'], timeout=5)
except subprocess.TimeoutExpired as e:
    print(e)

運行結(jié)果如下：

hello timeout
type(r)=<class 'subprocess.CompletedProcess'>,
    r.args=['echo', 'hello timeout'],
    r.returncode=0,
    r.stdout=None,
    r.stderr=None
PING www.google.com (216.58.194.164) 56(84) bytes of data.
64 bytes from ...: icmp_seq=1 ttl=54 time=10.4 ms
64 bytes from ...: icmp_seq=2 ttl=54 time=5.90 ms
64 bytes from ...: icmp_seq=3 ttl=54 time=6.19 ms
64 bytes from ...: icmp_seq=4 ttl=54 time=9.04 ms
64 bytes from ...: icmp_seq=5 ttl=54 time=16.7 ms
Command '['ping', 'www.google.com']' timed out after 5 seconds

當(dāng)超時時，會拋出一個 TimeoutExpired 異常。

信號（Signals）

對于 UNIX 系統(tǒng)，還可以使用 signal 模塊，通過在 5 秒后向處理器發(fā)送信號來引發(fā)異常。不過，這種方法相對底層且不夠直觀。

import signal
def handler(signum, frame):
    raise TimeoutError("函數(shù)超時")
def my_function():
    pass
signal.signal(signal.SIGALRM, handler)
signal.alarm(5)
try:
    my_function()
except TimeoutError:
    print("函數(shù)超時")
finally:
    signal.alarm(0)

知識補充

Python函數(shù)超時后不拋出異常的方法

1.func_timeout庫的使用及異常處理

func_timeout 庫簡介：使用python第三方 func_timeout 模塊中提供的 func_set_timeout 裝飾器可以非常簡單的設(shè)置python程序的超時時間。

import time
import func_timeout
from func_timeout import func_set_timeout

@func_set_timeout(3)
def task():
    while True:
        print('hello world')
        time.sleep(1)

if __name__ == '__main__':
    try:
        task()
    except func_timeout.exceptions.FunctionTimedOut:
        print("執(zhí)行已超時3秒")

上述代碼使用@func_set_timeout(3)，設(shè)置了函數(shù)超時時間為3秒，同時使用except func_timeout.exceptions.FunctionTimedOut: print("執(zhí)行已超時3秒")，進(jìn)行異常處理，輸出結(jié)果如下圖：

2.func_timeout阻塞主線程問題

什么是線程阻塞：在某一時刻某一個線程在運行一段代碼的時候，這時候另一個線程也需要運行，但是在運行過程中的那個線程執(zhí)行完成之前，另一個線程無法獲取到CPU執(zhí)行權(quán)，這時就會造成線程阻塞。

這里發(fā)現(xiàn)另外一個問題，鏈接中的如下代碼，其實還是會阻塞主線程。

import time
from func_timeout import func_set_timeout
?
@func_set_timeout(3)
def task():
    while True:
        print('hello world')
        time.sleep(1)
?
if __name__ == '__main__':
    task()

怎么驗證呢，我們在task()后面加一句print("a")，改成如下代碼：

import time
from func_timeout import func_set_timeout

@func_set_timeout(3)
def task():
    while True:
        print('hello world')
        time.sleep(1)

if __name__ == '__main__':
    task()
    print("a")

輸出結(jié)果如下：

從程序的輸出結(jié)果可以發(fā)現(xiàn)，控制臺并沒有打印字母a，所以按鏈接的代碼，還是會導(dǎo)致整個程序阻塞，等待task()執(zhí)行完畢，直到超時報錯。

3.啟動多線程解決阻塞問題

Python的多線程：Python提供兩個模塊進(jìn)行多線程的操作，分別是thread和threading，前者是比較低級的模塊，用于更底層的操作，一般應(yīng)用級別的開發(fā)不常用。

為了解決阻塞主線程的問題，這里嘗試使用threading庫，啟動多線程來執(zhí)行task()，代碼如下：

import time
from func_timeout import func_set_timeout
from threading import Thread

@func_set_timeout(3)
def task():
    while True:
        print('hello world')
        time.sleep(1)

if __name__ == '__main__':
    t1 = Thread(target=task, args=())
    t1.start()
    print("a")

程序運行結(jié)果，如下圖：

從程序輸出的結(jié)果可以看出，現(xiàn)在程序并沒有阻塞主線程，如期在控制臺上打印了字母a。同時，在等待3秒后，task()函數(shù)超時拋出了超時異常。

4.try語句封裝函數(shù)，實現(xiàn)全程捕獲異常

異常處理是編程語言或計算機(jī)硬件里的一種機(jī)制，用于處理軟件或信息系統(tǒng)中出現(xiàn)的異常狀況（即超出程序正常執(zhí)行流程的某些特殊條件）。

到這里，其實還是有一個問題沒有解決。那就是剛才程序輸出的異常還沒被捕獲和處理，為了得到完整的解決方案，代碼中還需要加上異常處理的部分，同時還需要另外寫一個函數(shù)try_task()，實現(xiàn)全程捕獲異常。

代碼如下：

import time
import func_timeout
from func_timeout import func_set_timeout
from threading import Thread


def try_task():
    try:
        task()
    except func_timeout.exceptions.FunctionTimedOut:
        print("執(zhí)行已超時3秒")


@func_set_timeout(3)
def task():
    while True:
        print('hello world')
        time.sleep(1)
      

if __name__ == '__main__':
    t1 = Thread(target=try_task, args=())
    t1.start()
    print("a")

最終程序輸出結(jié)果如下圖，實現(xiàn)了不阻塞主線程，3秒后超時拋出異常并被正常捕獲。

總結(jié)

在開發(fā)中，限制函數(shù)執(zhí)行時間是提升程序穩(wěn)定性和用戶體驗的重要手段。本文介紹了幾種實現(xiàn)方法：

• func-timeout 庫：通過 func_timeout 或裝飾器 func_set_timeout，可為函數(shù)設(shè)置超時時間，超時則拋出異常。適用于網(wǎng)絡(luò)請求或計算密集型任務(wù)。
• 自定義進(jìn)程：利用 multiprocessing 模塊創(chuàng)建子進(jìn)程執(zhí)行函數(shù)，通過 join(timeout) 控制超時，超時后終止進(jìn)程。
• subprocess 模塊：從 Python 3.5 起，subprocess.run() 支持超時參數(shù)，超時會拋出 TimeoutExpired 異常，適合外部命令調(diào)用。
• 信號機(jī)制：在 UNIX 系統(tǒng)中，使用 signal 模塊設(shè)置超時信號，超時后觸發(fā)異常，但實現(xiàn)較底層。

這些方法各有優(yōu)劣，開發(fā)者可根據(jù)實際需求選擇合適的方案。

到此這篇關(guān)于Python處理函數(shù)調(diào)用超時方法的詳細(xì)教學(xué)的文章就介紹到這了,更多相關(guān)Python處理函數(shù)調(diào)用超時內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評論