Python处理函数调用超时方法的详细教学

pip install func-timeout

from func_timeout import func_timeout, FunctionTimedOut
import time

def long_running_task():
    time.sleep(5)  # 模拟长时间运行的任务
    return "Task completed"
try:
    result = func_timeout(3, long_running_task)  # 设置3秒超时
    print(result)
except FunctionTimedOut:
    print("Function execution timed out!")

from func_timeout import func_set_timeout
import time

@func_set_timeout(2)  # 限制该函数的运行时间为2秒
def long_task():
    time.sleep(5)  # 任务实际需要5秒
    return "Finished"

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

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!')

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)

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

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("执行已超时3秒")

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

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

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

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("执行已超时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")