快捷導(dǎo)航

Python使用asyncio異步時(shí)的常見問(wèn)題總結(jié)

更新時(shí)間：2023年04月17日 14:09:44 作者：冷凍工廠

這篇文章主要為大家整理了開發(fā)人員在?Python?中使用?asyncio?時(shí)提出的常見問(wèn)題以及解決方法，文中的示例代碼講解詳細(xì)，感興趣的可以學(xué)習(xí)一下

1. 如何停止任務(wù)？

我們可以通過(guò) asyncio.Task 對(duì)象上的 cancel() 方法取消任務(wù)。如果任務(wù)被取消，cancel() 方法返回 True，否則返回 False。

...
# cancel the task
was_cancelled = task.cancel()

如果任務(wù)已經(jīng)完成，則無(wú)法取消，cancel() 方法將返回 False，任務(wù)不會(huì)處于已取消狀態(tài)。

下次任務(wù)有機(jī)會(huì)運(yùn)行時(shí)，它將引發(fā) CancelledError 異常。如果 CancelledError 異常未在包裝協(xié)程內(nèi)處理，任務(wù)將被取消。

否則，如果在包裝協(xié)程內(nèi)處理了 CancelledError 異常，任務(wù)將不會(huì)被取消。cancel() 方法還可以接受一個(gè)消息參數(shù)，該參數(shù)將在 CancelledError 的內(nèi)容中使用。

我們可以探索如何取消正在運(yùn)行的任務(wù)。

在這個(gè)例子中，我們定義了一個(gè)任務(wù)協(xié)程，它報(bào)告一條消息然后阻塞片刻。

然后我們定義用作 asyncio 程序入口點(diǎn)的主協(xié)程。它報(bào)告一條消息，創(chuàng)建并安排任務(wù)，然后等待片刻。

然后主協(xié)程在運(yùn)行時(shí)恢復(fù)和取消任務(wù)。它再等一會(huì)兒，讓任務(wù)響應(yīng)取消請(qǐng)求。然后主協(xié)程報(bào)告取消任務(wù)的請(qǐng)求是否成功。

任務(wù)被取消，然后完成。主協(xié)程然后在關(guān)閉程序之前報(bào)告任務(wù)的狀態(tài)是否已取消。

# SuperFastPython.com
# example of canceling a running task
import asyncio
 
# define a coroutine for a task
async def task_coroutine():
    # report a message
    print('executing the task')
    # block for a moment
    await asyncio.sleep(1)
 
# custom coroutine
async def main():
    # report a message
    print('main coroutine started')
    # create and schedule the task
    task = asyncio.create_task(task_coroutine())
    # wait a moment
    await asyncio.sleep(0.1)
    # cancel the task
    was_cancelled = task.cancel()
    # report whether the cancel request was successful
    print(f'was canceled: {was_cancelled}')
    # wait a moment
    await asyncio.sleep(0.1)
    # check the status of the task
    print(f'canceled: {task.cancelled()}')
    # report a final message
    print('main coroutine done')
 
# start the asyncio program
asyncio.run(main())

運(yùn)行該示例會(huì)啟動(dòng) asyncio 事件循環(huán)并執(zhí)行 main() 協(xié)程。main() 協(xié)程報(bào)告一條消息，然后創(chuàng)建并調(diào)度任務(wù)協(xié)程。然后它暫停并等待片刻以允許任務(wù)協(xié)程開始運(yùn)行。任務(wù)運(yùn)行，報(bào)告消息并休眠一段時(shí)間。

main() 協(xié)程恢復(fù)和取消任務(wù)。它報(bào)告取消任務(wù)的請(qǐng)求已成功。然后它會(huì)休眠片刻，讓任務(wù)響應(yīng)要取消的請(qǐng)求。

task_coroutine() 恢復(fù)并引發(fā) CancelledError 異常，導(dǎo)致任務(wù)失敗并完成。main()協(xié)程恢復(fù)并報(bào)告任務(wù)是否處于取消狀態(tài)。在這種情況下，確實(shí)如此。

此示例突出顯示了取消正在運(yùn)行的任務(wù)的正常情況。

main coroutine started
executing the task
was canceled: True
canceled: True
main coroutine done

2. 如何等待任務(wù)完成？

我們可以通過(guò)直接等待 asyncio.Task 對(duì)象來(lái)等待任務(wù)完成。

...
# wait for the task to finish
await task

我們可以在一行中創(chuàng)建和等待任務(wù)。

...
# create and wait for the task to finish
await asyncio.create_task(custom_coro())

3. 如何從任務(wù)中獲取返回值？

我們可能需要將協(xié)程的值返回給調(diào)用者。我們可以通過(guò)等待從協(xié)程中檢索返回值。它假定正在等待的另一個(gè)協(xié)程返回一個(gè)值。

# coroutine that returns a value
async def other_coro():
	return 100

等待其他協(xié)程將掛起調(diào)用協(xié)程并安排其他協(xié)程執(zhí)行。一旦另一個(gè)協(xié)程完成，調(diào)用協(xié)程將恢復(fù)。返回值將從另一個(gè)協(xié)程傳遞給調(diào)用者。

...
# execute coroutine and retrieve return value
value = await other_coro()

協(xié)程可以包裝在 asyncio.Task 對(duì)象中。這有助于獨(dú)立執(zhí)行協(xié)程而無(wú)需當(dāng)前協(xié)程等待它。

這可以使用 asyncio.create_task() 函數(shù)來(lái)實(shí)現(xiàn)。

...
# wrap coroutine in a task and schedule it for execution
task = asyncio.create_task(other_coro())

有兩種方法可以從 asyncio.Task 中檢索返回值，它們是：

等待任務(wù)。
調(diào)用結(jié)果() 方法。

我們可以等待任務(wù)來(lái)檢索返回值。如果任務(wù)已安排或正在運(yùn)行，則調(diào)用者將掛起，直到任務(wù)完成并提供返回值。如果任務(wù)完成，將立即提供返回值。

...
# get the return value from a task
value = await task

與協(xié)程不同，我們可以多次等待任務(wù)而不會(huì)引發(fā)錯(cuò)誤。

...
# get the return value from a task
value = await task
# get the return value from a task
value = await task

我們還可以通過(guò)調(diào)用 asyncio.Task 對(duì)象的 result() 方法來(lái)獲取任務(wù)的返回值。

...
# get the return value from a task
value = task.result()

這需要完成任務(wù)。如果不是，將引發(fā) InvalidStateError 異常。如果任務(wù)被取消，將引發(fā) CancelledError 異常。

4. 如何在后臺(tái)運(yùn)行任務(wù)？

我們可以通過(guò)將協(xié)程包裝在 asyncio.Task 對(duì)象中來(lái)在后臺(tái)運(yùn)行協(xié)程。這可以通過(guò)調(diào)用 asyncio.create_task() 函數(shù)并將其傳遞給協(xié)程來(lái)實(shí)現(xiàn)。

協(xié)程將被包裝在一個(gè) Task 對(duì)象中，并被安排執(zhí)行。將返回任務(wù)對(duì)象，調(diào)用者不會(huì)掛起。

...
# schedule the task for execution
task = asyncio.create_task(other_coroutine())

至少在當(dāng)前協(xié)程出于任何原因掛起之前，任務(wù)不會(huì)開始執(zhí)行。我們可以通過(guò)暫停片刻讓任務(wù)開始運(yùn)行來(lái)幫助解決問(wèn)題。這可以通過(guò)休眠零秒來(lái)實(shí)現(xiàn)。

...
# suspend for a moment to allow the task to start running
await asyncio.sleep(0)

這將暫停調(diào)用者一小會(huì)兒，并允許請(qǐng)求運(yùn)行的機(jī)會(huì)。這不是必需的，因?yàn)檎{(diào)用者可能會(huì)在未來(lái)某個(gè)時(shí)間暫停或作為正常執(zhí)行的一部分終止。一旦調(diào)用者沒有事情要做，我們也可以直接等待任務(wù)。

...
# wait for the task to complete
await task

5. 如何等待所有后臺(tái)任務(wù)？

我們可以等待 asyncio 程序中的所有獨(dú)立任務(wù)。這可以通過(guò)首先通過(guò) asyncio.all_tasks() 函數(shù)獲取一組所有當(dāng)前正在運(yùn)行的任務(wù)來(lái)實(shí)現(xiàn)。

...
# get a set of all running tasks
all_tasks = asyncio.all_tasks()

這將返回一個(gè)集合，其中包含一個(gè) asyncio.Task 對(duì)象，用于當(dāng)前正在運(yùn)行的每個(gè)任務(wù)，包括 main() 協(xié)程。

我們不能直接等待這個(gè)集合，因?yàn)樗鼤?huì)永遠(yuǎn)阻塞，因?yàn)樗?dāng)前任務(wù)。因此，我們可以獲取當(dāng)前正在運(yùn)行的任務(wù)的 asyncio.Task 對(duì)象并將其從集合中刪除。

這可以通過(guò)首先調(diào)用 asyncio.current_task() 方法來(lái)獲取當(dāng)前協(xié)程的任務(wù)，然后通過(guò) remove() 方法將其從集合中刪除來(lái)實(shí)現(xiàn)。

...
# get the current tasks
current_task = asyncio.current_task()
# remove the current task from the list of all tasks
all_tasks.remove(current_task)

最后，我們可以等待剩余的任務(wù)集。這將掛起調(diào)用者，直到集合中的所有任務(wù)都完成。

...
# suspend until all tasks are completed
await asyncio.wait(all_tasks)

將它們結(jié)合在一起，下面添加到 main() 協(xié)程末尾的代碼片段將等待所有后臺(tái)任務(wù)完成。

...
# get a set of all running tasks
all_tasks = asyncio.all_tasks()
# get the current tasks
current_task = asyncio.current_task()
# remove the current task from the list of all tasks
all_tasks.remove(current_task)
# suspend until all tasks are completed
await asyncio.wait(all_tasks)

到此這篇關(guān)于Python使用asyncio異步時(shí)的常見問(wèn)題總結(jié)的文章就介紹到這了,更多相關(guān)Python asyncio異步內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: