快捷導(dǎo)航

Python?異步如何使用等待有時(shí)間限制協(xié)程

更新時(shí)間：2023年03月22日 11:56:27 作者：冷凍工廠

這篇文章主要為大家介紹了Python?異步如何使用等待有時(shí)間限制協(xié)程示例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

正文

我們可以使用 asyncio.wait_for() 函數(shù)等待 asyncio 任務(wù)或協(xié)程超時(shí)完成。如果在任務(wù)完成之前超時(shí)已過，任務(wù)將被取消。

1. 什么是 Asyncio wait_for()

asyncio.wait_for() 函數(shù)允許調(diào)用者等待 asyncio 任務(wù)或協(xié)程超時(shí)完成。如果沒有指定超時(shí)，wait_for() 函數(shù)將等待直到任務(wù)完成。如果在任務(wù)完成之前指定了超時(shí)并超時(shí)，那么任務(wù)將被取消。

這允許調(diào)用者既可以設(shè)置他們?cè)敢獾却蝿?wù)完成的時(shí)間，又可以通過在超時(shí)結(jié)束時(shí)取消任務(wù)來強(qiáng)制執(zhí)行超時(shí)。

現(xiàn)在我們知道了 asyncio.wait_for() 函數(shù)是什么，讓我們看看如何使用它。

2. 如何使用 Asyncio wait_for()

asyncio.wait_for() 函數(shù)接受一個(gè)等待和超時(shí)。等待對(duì)象可能是協(xié)程或任務(wù)。必須指定超時(shí)，并且可以是無超時(shí)、整數(shù)或浮點(diǎn)秒數(shù)。wait_for() 函數(shù)返回一個(gè)協(xié)程，該協(xié)程在明確等待或作為任務(wù)調(diào)度之前不會(huì)執(zhí)行。

...
# wait for a task to complete
await asyncio.wait_for(coro, timeout=10)

如果提供協(xié)程，則在執(zhí)行 wait_for() 協(xié)程時(shí)將其轉(zhuǎn)換為任務(wù)。如果在任務(wù)完成之前超時(shí)已過，任務(wù)將被取消，并引發(fā) asyncio.TimeoutError，這可能需要處理。

...
# execute a task with a timeout
try:
	# wait for a task to complete
	await asyncio.wait_for(coro, timeout=1)
except asyncio.TimeoutError:
	# ...

如果等待的任務(wù)因未處理的異常而失敗，則該異常將傳播回等待 wait_for() 協(xié)程的調(diào)用者，在這種情況下可能需要處理它。

...
# execute a task that may fail
try:
	# wait for a task to complete
	await asyncio.wait_for(coro, timeout=1)
except asyncio.TimeoutError:
	# ...
except Exception:
	# ...

接下來，讓我們看看如何在超時(shí)時(shí)調(diào)用 wait_for()。

3. 帶有超時(shí)的 Asyncio wait_for() 示例

我們可以探索如何在任務(wù)完成之前等待具有超時(shí)的協(xié)程。在此示例中，我們執(zhí)行上述協(xié)程，但調(diào)用方等待 0.2 秒或 200 毫秒的固定超時(shí)?；叵胍幌?，一秒等于 1,000 毫秒。

任務(wù)協(xié)程被修改，使其休眠一秒以上，確保超時(shí)總是在任務(wù)完成之前到期。

# SuperFastPython.com
# example of waiting for a coroutine with a timeout
from random import random
import asyncio
# coroutine to execute in a new task
async def task_coro(arg):
    # generate a random value between 0 and 1
    value = 1 + random()
    # report message
    print(f'>task got {value}')
    # block for a moment
    await asyncio.sleep(value)
    # report all done
    print('>task done')
# main coroutine
async def main():
    # create a task
    task = task_coro(1)
    # execute and wait for the task without a timeout
    try:
        await asyncio.wait_for(task, timeout=0.2)
    except asyncio.TimeoutError:
        print('Gave up waiting, task canceled')
# start the asyncio program
asyncio.run(main())

運(yùn)行示例首先創(chuàng)建 main() 協(xié)程并將其用作 asyncio 程序的入口點(diǎn)。main() 協(xié)程創(chuàng)建任務(wù)協(xié)程。然后它調(diào)用 wait_for() 并傳遞任務(wù)協(xié)程并將超時(shí)設(shè)置為 0.2 秒。

main()協(xié)程被掛起，執(zhí)行task_coro()。它報(bào)告一條消息并休眠片刻。main() 協(xié)程在超時(shí)結(jié)束后恢復(fù)。 wait_for()協(xié)程取消task_coro()協(xié)程，main()協(xié)程掛起。

task_coro() 再次運(yùn)行并響應(yīng)要終止的請(qǐng)求。它引發(fā) TimeoutError 異常并終止。main() 協(xié)程恢復(fù)并處理由 task_coro() 引發(fā)的 TimeoutError。

這突出顯示了我們?nèi)绾握{(diào)用帶超時(shí)的 wait_for() 函數(shù)，并在任務(wù)未在超時(shí)內(nèi)完成時(shí)取消任務(wù)。

由于使用了隨機(jī)數(shù)，程序每次運(yùn)行時(shí)的輸出都會(huì)不同。