快捷導(dǎo)航

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

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

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

前言

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

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

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

func-timeout

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

1. 安裝 func-timeout

可以使用 pip 安裝：

pip install func-timeout

2. 基本用法

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

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

解釋：

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

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

from func_timeout import func_set_timeout
import time
 
@func_set_timeout(2)  # 限制該函數(shù)的運(yùn)行時(shí)間為2秒
def long_task():
    time.sleep(5)  # 任務(wù)實(shí)際需要5秒
    return "Finished"
 
try:
    print(long_task())
except FunctionTimedOut:
    print("Function execution timed out!")

這種方式適用于需要多次調(diào)用的函數(shù)，避免每次調(diào)用都手動(dòng)設(shè)置超時(shí)。

func-timeout 本質(zhì)上還是依賴多線程或多進(jìn)程實(shí)現(xiàn)超時(shí)控制，在某些情況下可能不適用于主線程（如 Jupyter Notebook）。它也不能用于 main 線程內(nèi)的 while True 死循環(huán)，因?yàn)?Python 的 GIL 可能會(huì)影響信號(hào)處理。

自定義進(jìn)程

除了使用上面的庫(kù)，也可以自己使用一個(gè)進(jìn)程來(lái)計(jì)時(shí)和檢測(cè)超時(shí)，另一個(gè)進(jìn)程來(lái)調(diào)用 Python 函數(shù)。以下是具體實(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 是一個(gè)無(wú)限迭代器
    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!')

運(yùn)行結(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ù)超時(shí)了（退出碼為 None），而 return_zero() 函數(shù)在 5 秒內(nèi)成功完成。

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

從 Python 3.5 開始，subprocess 模塊提供了一個(gè)便捷且推薦使用的 run() API，它內(nèi)置了超時(shí)支持。以下是示例代碼：

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)

運(yùn)行結(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)超時(shí)時(shí)，會(huì)拋出一個(gè) TimeoutExpired 異常。

信號(hào)（Signals）

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

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

總結(jié)

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

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