限制函數(shù)調(diào)用的最大時(shí)間是一種非常實(shí)用的技術(shù)手段，能夠幫助開(kāi)發(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 開(kāi)始，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)

知識(shí)補(bǔ)充

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

1.func_timeout庫(kù)的使用及異常處理

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

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í)行已超時(shí)3秒")

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

2.func_timeout阻塞主線程問(wèn)題

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

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

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()

怎么驗(yàn)證呢，我們?cè)?code>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)，控制臺(tái)并沒(méi)有打印字母a，所以按鏈接的代碼，還是會(huì)導(dǎo)致整個(gè)程序阻塞，等待task()執(zhí)行完畢，直到超時(shí)報(bào)錯(cuò)。

3.啟動(dòng)多線程解決阻塞問(wèn)題

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

為了解決阻塞主線程的問(wèn)題，這里嘗試使用threading庫(kù)，啟動(dòng)多線程來(lái)執(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")

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

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

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

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

到這里，其實(shí)還是有一個(gè)問(wèn)題沒(méi)有解決。那就是剛才程序輸出的異常還沒(méi)被捕獲和處理，為了得到完整的解決方案，代碼中還需要加上異常處理的部分，同時(shí)還需要另外寫一個(gè)函數(shù)try_task()，實(shí)現(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í)行已超時(shí)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é)果如下圖，實(shí)現(xiàn)了不阻塞主線程，3秒后超時(shí)拋出異常并被正常捕獲。

總結(jié)

在開(kāi)發(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)較底層。

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

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

最新評(píng)論