import signal


def handler(signum, frame):
    """
    定義信號處理函數(shù) handler，當接收到一個信號時，這個函數(shù)會被調用。在函數(shù)內部，拋出 TimeoutError 異常，表示操作超時。

    signum：表示接收到的信號編號（在本例中是 SIGALRM 信號）。
    frame：是當前的堆棧幀信息，我們在這個示例中并不使用它。
    當 handler 被調用時，它會拋出一個 TimeoutError 異常，表示任務超時。
    """
    raise TimeoutError("操作超時")


signal.signal(signal.SIGALRM, handler)
# 通過 signal.signal() 函數(shù)將 SIGALRM 信號與 handler 關聯(lián)。即：當程序接收到 SIGALRM 信號時，會執(zhí)行 handler 函數(shù)。
# SIGALRM 是一個定時信號，通常用于指定某個操作的超時時間。在這個例子中，下面我會設置一個定時器，讓它在 5 秒后發(fā)送 SIGALRM 信號。
signal.alarm(5)
# 設置一個定時器，定時器將在 5 秒后向當前進程發(fā)出一個 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)
# 創(chuàng)建一個定時器 t2，它會在 5 秒后執(zhí)行 timeout_fun() 函數(shù)。如果任務在 5 秒內未完成，timeout_fun 就會被調用，打印超時信息。

t1.start()
t2.start()

t1.join()
# 會讓主線程等待 t1 線程完成執(zhí)行。如果 t1 在 10 秒內完成，主線程才會繼續(xù)執(zhí)行下一步。如果 t1 超過了 10 秒，主線程會繼續(xù)等待。
t2.cancel()  
# 取消定時器 t2，如果 t1 任務在 5 秒內完成，定時器 t2 會被取消，防止超時處理函數(shù) timeout_fun() 被觸發(fā)。

import concurrent.futures
import time


def task():
    time.sleep(10)  # 模擬長時間任務
    return "任務完成"


with concurrent.futures.ThreadPoolExecutor() as executor:
    future = executor.submit(task)
    # submit() 方法異步提交任務并返回一個 Future 對象，F(xiàn)uture 對象提供了管理任務的接口，比如獲取任務結果或設置超時等。
    try:
        result = future.result(timeout=5)
        # 用來獲取 task() 函數(shù)的執(zhí)行結果。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()，但它是非阻塞的，因此可以在等待時繼續(xù)執(zhí)行其他任務。
    return "任務完成"


async def main():
    try:
        result = await asyncio.wait_for(task(), timeout=5)
        # asyncio.wait_for() 用來執(zhí)行 task() 并設置超時。如果 task() 在指定的時間內（這里是 5 秒）完成，wait_for 會返回 task() 的結果（即 "任務完成"）。
        print(result)
    except asyncio.TimeoutError:
        print("任務超時")


asyncio.run(main())
# 用來啟動事件循環(huán)并執(zhí)行 main() 協(xié)程。asyncio.run() 會創(chuàng)建一個新的事件循環(huán)，運行傳入的協(xié)程，直到協(xié)程執(zhí)行完畢并返回結果。這個方法會自動管理事件循環(huán)的創(chuàng)建和關閉。

import eventlet


def long_task():
    eventlet.sleep(10)
    # 通過 eventlet.sleep(10) 模擬一個阻塞的操作，實際上是讓出當前綠色線程 10 秒鐘。
    # 注意：eventlet.sleep() 并不是真的讓程序進入休眠，它讓出控制權，使得其他綠色線程可以執(zhí)行。這是 eventlet 模擬并發(fā)的方式。
    return "任務完成"


def task_with_timeout():
    pool = eventlet.GreenPool()  # 創(chuàng)建綠色線程池
    result = pool.spawn(long_task)
    # spawn() 方法會異步地執(zhí)行 long_task，并返回一個 GreenThread 對象。這個對象表示正在運行的綠色線程。
    try:
        # wait() 方法會阻塞當前線程，直到綠色線程執(zhí)行完畢并返回結果。它的 timeout 參數(shù)用于設置最大等待時間，這里設置為 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("任務超時")