快捷導航

Python處理函數(shù)調(diào)用超時的四種方法

更新時間：2025年04月20日 14:52:09 作者：呱牛?do?IT

在實際開發(fā)過程中,我們可能會遇到一些場景,需要對函數(shù)的執(zhí)行時間進行限制,例如,當一個函數(shù)執(zhí)行時間過長時,可能會導致程序卡頓、資源占用過高,因此,在某些情況下,我們希望限制函數(shù)調(diào)用的最大時所以本文給大家介紹了Python處理函數(shù)調(diào)用超時的四種方法

前言

在實際開發(fā)過程中，我們可能會遇到一些場景，需要對函數(shù)的執(zhí)行時間進行限制。例如，當一個函數(shù)執(zhí)行時間過長時，可能會導致程序卡頓、資源占用過高，甚至影響整個系統(tǒng)的穩(wěn)定性。因此，在某些情況下，我們希望限制函數(shù)調(diào)用的最大時間，以確保程序能夠在合理的時間范圍內(nèi)完成任務(wù)，或者在超時的情況下采取其他措施。

為了實現(xiàn)這一目標，可以通過多種方式來控制函數(shù)的執(zhí)行時間。例如，可以使用多線程或異步編程技術(shù)，在指定的時間范圍內(nèi)監(jiān)控函數(shù)的執(zhí)行情況。如果函數(shù)在規(guī)定時間內(nèi)未能完成執(zhí)行，則可以中斷該函數(shù)的運行，并返回一個超時提示或執(zhí)行備用邏輯。這種方式不僅能夠提高程序的健壯性，還能有效避免因單個函數(shù)執(zhí)行時間過長而導致的系統(tǒng)性能問題。

限制函數(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ì)上還是依賴多線程或多進程實現(xiàn)超時控制，在某些情況下可能不適用于主線程（如 Jupyter Notebook）。它也不能用于 main 線程內(nèi)的 while True 死循環(huán)，因為 Python 的 GIL 可能會影響信號處理。

自定義進程

除了使用上面的庫，也可以自己使用一個進程來計時和檢測超時，另一個進程來調(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

當超時時，會拋出一個 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)

總結(jié)

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

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