Python異步之在Asyncio中怎么運行阻塞任務

本文講解"Python異步之在Asyncio中如何運行阻塞任務"，希望能夠解決相關問題。

正文

阻塞任務是阻止當前線程繼續進行的任務。

如果在 asyncio 程序中執行阻塞任務，它會停止整個事件循環，從而阻止任何其他協程繼續進行。

我們可以通過 asyncio.to_thread() 和 loop.run_in_executor() 函數在 asyncio 程序中異步運行阻塞調用。

1. 阻塞任務

asyncio的重點是異步編程和非阻塞IO。然而，我們經常需要在 asyncio 應用程序中執行阻塞函數調用。

這可能有很多原因，例如：

• 執行 CPU 密集型任務，例如計算某事。

• 執行阻塞 IO 綁定任務，如從文件讀取或寫入。

• 調用不支持 asyncio 的第三方庫。

直接在 asyncio 程序中進行阻塞調用將導致事件循環在執行阻塞調用時停止。它不允許其他協程在后臺運行。

我們如何在 asyncio 程序中異步執行阻塞調用？

2. 如何運行阻塞任務

asyncio 模塊提供了兩種在 asyncio 程序中執行阻塞調用的方法。

第一種是使用 asyncio.to_thread() 函數。這是在高級 API 中，供應用程序開發人員使用。

asyncio.to_thread() 函數采用要執行的函數名和任何參數。

該函數在單獨的線程中執行。它返回一個可以作為獨立任務等待或安排的協程。

...
# execute a function in a separate thread
await asyncio.to_thread(task)

在返回的協程有機會在事件循環中運行之前，任務不會開始執行。asyncio.to_thread() 函數在后臺創建一個 ThreadPoolExecutor 來執行阻塞調用。因此，asyncio.to_thread() 函數僅適用于 IO 綁定任務。

另一種方法是使用 loop.run_in_executor() 函數。

這是在低級異步 API 中，首先需要訪問事件循環，例如通過 asyncio.get_running_loop() 函數。

loop.run_in_executor() 函數接受一個執行器和一個要執行的函數。

如果沒有為執行器提供，則使用默認執行器，即 ThreadPoolExecutor。

loop.run_in_executor() 函數返回一個可等待對象，如果需要可以等待它。任務將立即開始執行，因此返回的可等待對象不需要等待或安排阻塞調用開始執行。

...
# get the event loop
loop = asyncio.get_running_loop()
# execute a function in a separate thread
await loop.run_in_executor(None, task)

或者，可以創建一個執行器并將其傳遞給 loop.run_in_executor() 函數，該函數將在執行器中執行異步調用。

在這種情況下，調用者必須管理執行器，一旦調用者完成它就將其關閉。

...
# create a process pool
with ProcessPoolExecutor as exe:
	# get the event loop
	loop = asyncio.get_running_loop()
	# execute a function in a separate thread
	await loop.run_in_executor(exe, task)
	# process pool is shutdown automatically...

這兩種方法允許阻塞調用作為異步任務在 asyncio 程序中執行。

現在我們知道如何在 asyncio 程序中執行阻塞調用，讓我們看一些有效的例子。

3. 實例

我們可以探索如何使用 asyncio.to_thread() 在 asyncio 程序中執行阻塞 IO 綁定調用。

在這個例子中，我們將定義一個函數來阻塞調用者幾秒鐘。然后，我們將使用 asyncio.to_thread() 函數在 asyncio 的線程池中異步執行此函數。

這將使呼叫者騰出時間繼續其他活動。

# SuperFastPython.com
# example of running a blocking io-bound task in asyncio
import asyncio
import time
# a blocking io-bound task
def blocking_task():
    # report a message
    print('Task starting')
    # block for a while
    time.sleep(2)
    # report a message
    print('Task done')
# main coroutine
async def main():
    # report a message
    print('Main running the blocking task')
    # create a coroutine for  the blocking task
    coro = asyncio.to_thread(blocking_task)
    # schedule the task
    task = asyncio.create_task(coro)
    # report a message
    print('Main doing other things')
    # allow the scheduled task to start
    await asyncio.sleep(0)
    # await the task
    await task
# run the asyncio program
asyncio.run(main())

運行示例首先創建 main() 協程并將其作為 asyncio 程序的入口點運行。main() 協程運行并報告一條消息。然后它發出對線程池的阻塞函數調用的調用。然后將協程包裝在任務中并獨立執行。

main() 協程可以自由地繼續其他活動。在這種情況下，它會休眠片刻以允許計劃任務開始執行。這使得目標函數可以在后臺下發給 ThreadPoolExecutor 并開始運行。

然后 main() 協程掛起并等待任務完成。阻塞函數報告一條消息，休眠 2 秒，然后報告最后一條消息。

這突出了我們如何在一個單獨的線程中與 asyncio 程序異步執行阻塞 IO 綁定任務。

Main running the blocking task
Main doing other things
Task starting
Task done

關于 "Python異步之在Asyncio中如何運行阻塞任務" 就介紹到此。希望多多支持碩編程。