import signal


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

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


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

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

import threading
import time


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


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


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

t1.start()
t2.start()

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

import concurrent.futures
import time


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


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

import asyncio


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


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


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

import eventlet


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


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


print(task_with_timeout())

from func_timeout import func_timeout, FunctionTimedOut


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


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