import signal


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

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


signal.signal(signal.SIGALRM, handler)
# 通過 signal.signal() 函數(shù)將 SIGALRM 信號與 handler 關(guān)聯(lián)。即：當(dāng)程序接收到 SIGALRM 信號時，會執(zhí)行 handler 函數(shù)。
# SIGALRM 是一個定時信號，通常用于指定某個操作的超時時間。在這個例子中，下面我會設(shè)置一個定時器，讓它在 5 秒后發(fā)送 SIGALRM 信號。
signal.alarm(5)
# 設(shè)置一個定時器，定時器將在 5 秒后向當(dāng)前進(jìn)程發(fā)出一個 SIGALRM 信號。

try:
    while True:  # 模擬長時間運(yùn)行的任務(wù)
        pass
except TimeoutError:
    print("操作超時")
finally:
    signal.alarm(0)  # 取消定時器，防止程序結(jié)束后定時器仍然生效。

import threading
import time


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


def timeout_fun():
    print("任務(wù)超時")


timeout_time = 5
t1 = threading.Thread(target=task)
t2 = threading.Timer(timeout_time, timeout_fun)
# 創(chuàng)建一個定時器 t2，它會在 5 秒后執(zhí)行 timeout_fun() 函數(shù)。如果任務(wù)在 5 秒內(nèi)未完成，timeout_fun 就會被調(diào)用，打印超時信息。

t1.start()
t2.start()

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

import concurrent.futures
import time


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


with concurrent.futures.ThreadPoolExecutor() as executor:
    future = executor.submit(task)
    # submit() 方法異步提交任務(wù)并返回一個 Future 對象，F(xiàn)uture 對象提供了管理任務(wù)的接口，比如獲取任務(wù)結(jié)果或設(shè)置超時等。
    try:
        result = future.result(timeout=5)
        # 用來獲取 task() 函數(shù)的執(zhí)行結(jié)果。timeout=5 表示最多等待 5 秒鐘，如果在 5 秒內(nèi)任務(wù)沒有完成，就會拋出 TimeoutError 異常。
        # 如果 task() 在 5 秒內(nèi)完成，future.result() 會返回任務(wù)的返回值（即 "任務(wù)完成"），然后打印出結(jié)果。
        print(result)
    except concurrent.futures.TimeoutError:
        print("任務(wù)超時")

import asyncio


async def task():
    await asyncio.sleep(10)
    # 模擬一個長時間運(yùn)行的異步任務(wù)，asyncio.sleep(10) 會讓當(dāng)前任務(wù)掛起 10 秒，在這段時間內(nèi)不會占用 CPU 資源。類似于 time.sleep()，但它是非阻塞的，因此可以在等待時繼續(xù)執(zhí)行其他任務(wù)。
    return "任務(wù)完成"


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


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

import eventlet


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


def task_with_timeout():
    pool = eventlet.GreenPool()  # 創(chuàng)建綠色線程池
    result = pool.spawn(long_task)
    # spawn() 方法會異步地執(zhí)行 long_task，并返回一個 GreenThread 對象。這個對象表示正在運(yùn)行的綠色線程。
    try:
        # wait() 方法會阻塞當(dāng)前線程，直到綠色線程執(zhí)行完畢并返回結(jié)果。它的 timeout 參數(shù)用于設(shè)置最大等待時間，這里設(shè)置為 5 秒。如果綠色線程在 5 秒內(nèi)沒有完成，wait() 方法會拋出 eventlet.timeout.Timeout 異常。
        return result.wait(timeout=5)  # 設(shè)置超時時間為5秒
    except eventlet.timeout.Timeout:
        return "任務(wù)超時"


print(task_with_timeout())

from func_timeout import func_timeout, FunctionTimedOut


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


try:
    result = func_timeout(5, long_task)  # 設(shè)置超時時間為5秒
    print(result)
except FunctionTimedOut:
    print("任務(wù)超時")