Python中六大超时处理方法介绍

import signal


def handler(signum, frame):
    """
    定义信号处理函数 handler，当接收到一个信号时，这个函数会被调用。在函数内部，抛出 TimeoutError 异常，表示操作超时。

    signum：表示接收到的信号编号（在本例中是 SIGALRM 信号）。
    frame：是当前的堆栈帧信息，我们在这个示例中并不使用它。
    当 handler 被调用时，它会抛出一个 TimeoutError 异常，表示任务超时。
    """
    raise TimeoutError("操作超时")


signal.signal(signal.SIGALRM, handler)
# 通过 signal.signal() 函数将 SIGALRM 信号与 handler 关联。即：当程序接收到 SIGALRM 信号时，会执行 handler 函数。
# SIGALRM 是一个定时信号，通常用于指定某个操作的超时时间。在这个例子中，下面我会设置一个定时器，让它在 5 秒后发送 SIGALRM 信号。
signal.alarm(5)
# 设置一个定时器，定时器将在 5 秒后向当前进程发出一个 SIGALRM 信号。

try:
    while True:  # 模拟长时间运行的任务
        pass
except TimeoutError:
    print("操作超时")
finally:
    signal.alarm(0)  # 取消定时器，防止程序结束后定时器仍然生效。

import threading
import time


def task():
    time.sleep(10)  # 模拟长时间任务
    return "任务完成"


def timeout_fun():
    print("任务超时")


timeout_time = 5
t1 = threading.Thread(target=task)
t2 = threading.Timer(timeout_time, timeout_fun)
# 创建一个定时器 t2，它会在 5 秒后执行 timeout_fun() 函数。如果任务在 5 秒内未完成，timeout_fun 就会被调用，打印超时信息。

t1.start()
t2.start()

t1.join()
# 会让主线程等待 t1 线程完成执行。如果 t1 在 10 秒内完成，主线程才会继续执行下一步。如果 t1 超过了 10 秒，主线程会继续等待。
t2.cancel()  
# 取消定时器 t2，如果 t1 任务在 5 秒内完成，定时器 t2 会被取消，防止超时处理函数 timeout_fun() 被触发。

import concurrent.futures
import time


def task():
    time.sleep(10)  # 模拟长时间任务
    return "任务完成"


with concurrent.futures.ThreadPoolExecutor() as executor:
    future = executor.submit(task)
    # submit() 方法异步提交任务并返回一个 Future 对象，Future 对象提供了管理任务的接口，比如获取任务结果或设置超时等。
    try:
        result = future.result(timeout=5)
        # 用来获取 task() 函数的执行结果。timeout=5 表示最多等待 5 秒钟，如果在 5 秒内任务没有完成，就会抛出 TimeoutError 异常。
        # 如果 task() 在 5 秒内完成，future.result() 会返回任务的返回值（即 "任务完成"），然后打印出结果。
        print(result)
    except concurrent.futures.TimeoutError:
        print("任务超时")

import asyncio


async def task():
    await asyncio.sleep(10)
    # 模拟一个长时间运行的异步任务，asyncio.sleep(10) 会让当前任务挂起 10 秒，在这段时间内不会占用 CPU 资源。类似于 time.sleep()，但它是非阻塞的，因此可以在等待时继续执行其他任务。
    return "任务完成"


async def main():
    try:
        result = await asyncio.wait_for(task(), timeout=5)
        # asyncio.wait_for() 用来执行 task() 并设置超时。如果 task() 在指定的时间内（这里是 5 秒）完成，wait_for 会返回 task() 的结果（即 "任务完成"）。
        print(result)
    except asyncio.TimeoutError:
        print("任务超时")


asyncio.run(main())
# 用来启动事件循环并执行 main() 协程。asyncio.run() 会创建一个新的事件循环，运行传入的协程，直到协程执行完毕并返回结果。这个方法会自动管理事件循环的创建和关闭。

import eventlet


def long_task():
    eventlet.sleep(10)
    # 通过 eventlet.sleep(10) 模拟一个阻塞的操作，实际上是让出当前绿色线程 10 秒钟。
    # 注意：eventlet.sleep() 并不是真的让程序进入休眠，它让出控制权，使得其他绿色线程可以执行。这是 eventlet 模拟并发的方式。
    return "任务完成"


def task_with_timeout():
    pool = eventlet.GreenPool()  # 创建绿色线程池
    result = pool.spawn(long_task)
    # spawn() 方法会异步地执行 long_task，并返回一个 GreenThread 对象。这个对象表示正在运行的绿色线程。
    try:
        # wait() 方法会阻塞当前线程，直到绿色线程执行完毕并返回结果。它的 timeout 参数用于设置最大等待时间，这里设置为 5 秒。如果绿色线程在 5 秒内没有完成，wait() 方法会抛出 eventlet.timeout.Timeout 异常。
        return result.wait(timeout=5)  # 设置超时时间为5秒
    except eventlet.timeout.Timeout:
        return "任务超时"


print(task_with_timeout())

from func_timeout import func_timeout, FunctionTimedOut


def long_task():
    import time
    time.sleep(10)  # 模拟长时间任务
    return "任务完成"


try:
    result = func_timeout(5, long_task)  # 设置超时时间为5秒
    print(result)
except FunctionTimedOut:
    print("任务超时")