Python異步之怎么保護(hù)任務(wù)免于取消

本文講解"Python異步之如何保護(hù)任務(wù)免于取消"，希望能夠解決相關(guān)問(wèn)題。

Asyncio 任務(wù)可以通過(guò)調(diào)用它們的 cancel() 方法來(lái)取消。我們可以通過(guò)將任務(wù)包裝在對(duì) asyncio.shield() 的調(diào)用中來(lái)保護(hù)任務(wù)不被取消。

1. 什么是 Asyncio shield()

asyncio.shield() 函數(shù)在 Future 中包裝了一個(gè)可等待對(duì)象，它將吸收要取消的請(qǐng)求。

這意味著被屏蔽的未來(lái)可以傳遞給可能?chē)L試取消它的任務(wù)，并且取消請(qǐng)求看起來(lái)像是成功的，除了被屏蔽的任務(wù)或協(xié)程將繼續(xù)運(yùn)行。

它可能在 asyncio 程序中很有用，其中某些任務(wù)可以取消，但其他任務(wù)（可能具有更高優(yōu)先級(jí)）則不能。

它也可能在某些任務(wù)可以安全取消的程序中很有用，例如那些在設(shè)計(jì)時(shí)考慮了 asyncio 的任務(wù)，而其他任務(wù)則不能安全終止，因此必須避免取消。

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

2. 如何使用 Asyncio shield()

asyncio.shield() 函數(shù)將保護(hù)另一個(gè)任務(wù)或協(xié)程不被取消。它以一個(gè)可等待對(duì)象作為參數(shù)并返回一個(gè) asyncio.Future 對(duì)象。

然后可以直接等待 Future 對(duì)象或?qū)⑵鋫鬟f給另一個(gè)任務(wù)或協(xié)程。

...
# shield a task from cancellation
shielded = asyncio.shield(task)
# await the shielded task
await shielded

返回的 Future 可以通過(guò)調(diào)用 cancel() 方法取消。

如果內(nèi)部任務(wù)正在運(yùn)行，請(qǐng)求將被報(bào)告為成功。

...
# cancel a shielded task
was_canceld = shielded.cancel()

任何等待 Future 對(duì)象的協(xié)程都會(huì)引發(fā) asyncio.CancelledError，這可能需要處理。

...
try:
	# await the shielded task
	await asyncio.shield(task)
except asyncio.CancelledError:
	# ...

重要的是，對(duì) Future 對(duì)象的取消請(qǐng)求不會(huì)傳播到內(nèi)部任務(wù)。這意味著取消請(qǐng)求被護(hù)盾吸收了。

...
# create a task
task = asyncio.create_task(coro())
# create a shield
shield = asyncio.shield(task)
# cancel the shield (does not cancel the task)
shield.cancel()

如果協(xié)程被提供給 asyncio.shield() 函數(shù)，它將被包裝在 asyncio.Task() 中并立即調(diào)度。

這意味著不需要等待屏蔽來(lái)讓內(nèi)部協(xié)程運(yùn)行。

如果被屏蔽的任務(wù)被取消，取消請(qǐng)求將向上傳播到屏蔽，屏蔽也將被取消。

...
# create a task
task = asyncio.create_task(coro())
# create a shield
shield = asyncio.shield(task)
# cancel the task (also cancels the shield)
task.cancel()

現(xiàn)在我們知道如何使用 asyncio.shield() 函數(shù)，讓我們看一些有效的例子。

3. 示例

我們可以探索如何使用 asyncio.shield() 來(lái)保護(hù)任務(wù)不被取消。

在這個(gè)例子中，我們定義了一個(gè)簡(jiǎn)單的協(xié)程任務(wù)，它接受一個(gè)整數(shù)參數(shù)，休眠一秒鐘，然后返回參數(shù)。然后可以創(chuàng)建協(xié)程并將其安排為任務(wù)。

我們可以定義第二個(gè)協(xié)程，它接受一個(gè)任務(wù)，休眠幾分之一秒，然后取消提供的任務(wù)。

在主協(xié)程中，我們可以屏蔽第一個(gè)任務(wù)，然后將其傳遞給第二個(gè)任務(wù)，然后等待被屏蔽的任務(wù)。

期望是屏蔽將被取消并保持內(nèi)部任務(wù)完好無(wú)損。取消將中斷主協(xié)程。我們可以在程序結(jié)束時(shí)檢查內(nèi)部任務(wù)的狀態(tài)，我們希望它已經(jīng)正常完成，而不管屏蔽上的取消請(qǐng)求如何。

# SuperFastPython.com
# example of using asyncio shield to protect a task from cancellation
import asyncio
# define a simple asynchronous
async def simple_task(number):
    # block for a moment
    await asyncio.sleep(1)
    # return the argument
    return number
# cancel the given task after a moment
async def cancel_task(task):
    # block for a moment
    await asyncio.sleep(0.2)
    # cancel the task
    was_cancelled = task.cancel()
    print(f'cancelled: {was_cancelled}')
# define a simple coroutine
async def main():
    # create the coroutine
    coro = simple_task(1)
    # create a task
    task = asyncio.create_task(coro)
    # created the shielded task
    shielded = asyncio.shield(task)
    # create the task to cancel the previous task
    asyncio.create_task(cancel_task(shielded))
    # handle cancellation
    try:
        # await the shielded task
        result = await shielded
        # report the result
        print(f'>got: {result}')
    except asyncio.CancelledError:
        print('shielded was cancelled')
    # wait a moment
    await asyncio.sleep(1)
    # report the details of the tasks
    print(f'shielded: {shielded}')
    print(f'task: {task}')
# start
asyncio.run(main())

運(yùn)行示例首先創(chuàng)建 main() 協(xié)程并將其用作應(yīng)用程序的入口點(diǎn)。創(chuàng)建任務(wù)協(xié)程，然后將其包裝并安排在任務(wù)中。然后該任務(wù)就不會(huì)被取消。

然后將屏蔽的任務(wù)傳遞給 cancel_task() 協(xié)程，該協(xié)程包裝在任務(wù)中并進(jìn)行調(diào)度。主協(xié)程然后等待受保護(hù)的任務(wù)，該任務(wù)需要 CancelledError 異常。

任務(wù)運(yùn)行片刻然后休眠。取消任務(wù)運(yùn)行片刻，休眠，恢復(fù)然后取消屏蔽任務(wù)。取消請(qǐng)求報(bào)告它已成功。

這會(huì)在受保護(hù)的 Future 中引發(fā) CancelledError 異常，但不會(huì)在內(nèi)部任務(wù)中引發(fā)。

main() 協(xié)程恢復(fù)并響應(yīng) CancelledError 異常，報(bào)告一條消息。然后它會(huì)睡一會(huì)兒。任務(wù)恢復(fù)、完成并返回一個(gè)值。

最后，main() 協(xié)程恢復(fù)，并報(bào)告被屏蔽的未來(lái)和內(nèi)部任務(wù)的狀態(tài)。我們可以看到屏蔽的未來(lái)被標(biāo)記為已取消，而內(nèi)部任務(wù)被標(biāo)記為正常完成并提供返回值。

此示例突出顯示了如何使用防護(hù)罩來(lái)成功保護(hù)內(nèi)部任務(wù)不被取消。

cancelled: True
shielded was cancelled
shielded: <future cancelled> task: <task?finished?name='task-2'?coro=