1 引言 

Python 的 asyncio 模塊為異步編程提供了強大的支持，但在某些場景下，我們可能需要處理異步任務(wù)與非異步（同步）任務(wù)的順序執(zhí)行或并行執(zhí)行。本篇文章將逐步帶你了解如何在 Python 中處理這些不同類型的任務(wù)。

2 順序執(zhí)行非異步任務(wù) 

在日常編程中，最常見的情況之一就是順序執(zhí)行一系列非異步（同步）的任務(wù)。這些任務(wù)在同一個線程中執(zhí)行，通常會阻塞主程序的運行，直到任務(wù)完成。

示例代碼：

import time

def blocking_task(id):
    print(f"Blocking task {id} started")
    time.sleep(2)  # 模擬一個阻塞操作
    print(f"Blocking task {id} finished")

# 順序執(zhí)行多個同步任務(wù)
def main():
    for i in range(3):
        blocking_task(i)

main()

解釋:

在此代碼中，每個任務(wù)按順序執(zhí)行，time.sleep() 會阻塞當前線程，直到所有任務(wù)結(jié)束。這種順序執(zhí)行的方式雖然簡單直接，但效率較低，尤其當任務(wù)涉及 I/O 操作時，會浪費大量時間。

3 順序執(zhí)行異步任務(wù)

如果我們希望提高任務(wù)的執(zhí)行效率，可以考慮使用異步任務(wù)。異步任務(wù)不會阻塞主線程，而是會等待特定的事件（例如 I/O 操作的完成），然后繼續(xù)執(zhí)行。

示例代碼：

import asyncio

async def async_task(id):
    print(f"Async task {id} started")
    await asyncio.sleep(2)  # 模擬異步操作
    print(f"Async task {id} finished")

# 順序執(zhí)行多個異步任務(wù)
async def main():
    for i in range(3):
        await async_task(i)

asyncio.run(main())

解釋：

通過使用 async def 定義異步函數(shù)，await 關(guān)鍵字用于暫停任務(wù)的執(zhí)行并等待異步操作完成。雖然這些任務(wù)是異步的，但由于我們使用了 await，它們?nèi)匀皇琼樞驁?zhí)行的。

3 并行執(zhí)行異步任務(wù)

在某些情況下，我們可能希望異步任務(wù)能夠并行執(zhí)行，而不是一個接一個地等待。此時可以使用 asyncio.gather()，它允許我們并行運行多個異步任務(wù)，從而提高程序效率。

示意圖：

示例代碼：

import asyncio

async def async_task(id):
    print(f"Async task {id} started")
    await asyncio.sleep(2)
    print(f"Async task {id} finished")

# 并行執(zhí)行多個異步任務(wù)
async def main():
    tasks = [async_task(i) for i in range(3)]
    await asyncio.gather(*tasks)

asyncio.run(main())

解釋：

asyncio.gather() 會并行執(zhí)行多個異步任務(wù)，而不是按順序等待。任務(wù)在后臺同時運行，極大提高了效率，尤其是當任務(wù)需要等待 I/O 時（例如網(wǎng)絡(luò)請求、文件操作等）。

? 注意：

tasks = [async_task(i) for i in range(3)] 這個時候不會執(zhí)行async_task函數(shù)，只是創(chuàng)建協(xié)程對象，還沒真正的啟動。

• 當你調(diào)用 async_task(i) 時，它不會立即執(zhí)行，而是返回一個 協(xié)程對象（coroutine object），這個對象代表一個等待執(zhí)行的異步任務(wù)。
• 只有當你 await 這個協(xié)程對象或者將它傳遞給 asyncio.gather()、asyncio.create_task()、asyncio.run() 等函數(shù)時，協(xié)程才會開始執(zhí)行。

4 并行執(zhí)行非異步任務(wù)（阻塞任務(wù)）

如果你有一些外部庫提供的阻塞任務(wù)（如文件讀寫、網(wǎng)絡(luò)操作等），這些任務(wù)無法直接變?yōu)楫惒胶瘮?shù)。為了與異步任務(wù)并行執(zhí)行這些阻塞任務(wù)，asyncio.run_in_executor() 是你的好幫手。

示意圖：

示例代碼：使用線程池并行執(zhí)行同步任務(wù)

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor

def blocking_task(id):
    print(f"Blocking task {id} started")
    time.sleep(2)
    print(f"Blocking task {id} finished")

async def main():
    with ThreadPoolExecutor() as pool:
        tasks = [
            asyncio.get_event_loop().run_in_executor(pool, blocking_task, i)
            for i in range(3)
        ]
        await asyncio.gather(*tasks)

asyncio.run(main())

解釋：

run_in_executor() 將阻塞的任務(wù)交給線程池（或進程池）執(zhí)行，而不會阻塞主事件循環(huán)。這使得我們可以同時處理異步任務(wù)和阻塞任務(wù)。

使用進程池還是線程池？

• 線程池（ThreadPoolExecutor）：適用于 I/O 密集型任務(wù)，如文件操作或網(wǎng)絡(luò)請求。這類任務(wù)通常會等待外部事件完成，因此不需要消耗大量 CPU 資源。
• 進程池（ProcessPoolExecutor）：適合 CPU 密集型任務(wù)，如數(shù)據(jù)處理和計算。使用進程池可以充分利用多核 CPU，提升性能。

示例代碼：使用進程池

from concurrent.futures import ProcessPoolExecutor
import asyncio

def cpu_intensive_task(id):
    print(f"CPU task {id} started")
    result = sum(i*i for i in range(10**6))  # 模擬CPU密集任務(wù)
    print(f"CPU task {id} finished with result: {result}")

async def main():
    with ProcessPoolExecutor() as pool:
        tasks = [
            asyncio.get_event_loop().run_in_executor(pool, cpu_intensive_task, i)
            for i in range(3)
        ]
        await asyncio.gather(*tasks)

asyncio.run(main())

5 總結(jié)

• 順序執(zhí)行非異步任務(wù)：通常用于簡單的任務(wù)，但效率低下，容易阻塞線程。
• 順序執(zhí)行異步任務(wù)：使用 asyncio 提供的異步函數(shù)，能夠在等待 I/O 時不阻塞主線程。
• 并行執(zhí)行異步任務(wù)：通過 asyncio.gather()，可以輕松并行多個異步任務(wù)，極大提高執(zhí)行效率。
• 并行執(zhí)行非異步任務(wù)：通過 run_in_executor() 將阻塞任務(wù)交給線程池或進程池，保證異步任務(wù)和同步任務(wù)可以并行執(zhí)行。
• 線程池 vs 進程池：選擇線程池處理 I/O 密集型任務(wù)，進程池處理 CPU 密集型任務(wù)。

通過這些技巧，你可以在 Python 中輕松管理各種類型的任務(wù)，實現(xiàn)高效并行處理。

到此這篇關(guān)于Python中asyncio的多種用法的文章就介紹到這了,更多相關(guān)Python中asyncio多種用法內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評論