高效能 Python：Asyncio-Python教學-PHP中文網

High-Performance Python: Asyncio

並發程式設計是一種處理多個任務同時執行的程式設計方法。在Python中，asyncio是實現非同步程式設計的強大工具。基於協程的概念，asyncio 可以有效率地處理 I/O 密集型任務。本文將介紹asyncio的基本原理與使用方法

High-Performance Python: Asyncio

為什麼我們需要 asyncio

我們知道，在處理I/O操作時，使用多執行緒相比普通的單執行緒可以大大提高效率。那麼，為什麼我們還需要 asyncio？

多執行緒有很多優點，應用廣泛，但也有一定的限制：

例如多執行緒的執行過程很容易被中斷，因此可能會出現race condition的情況。
而且，執行緒切換本身是有一定成本的，執行緒數不可能無限增加。因此，如果你的I/O操作非常繁重，多執行緒很可能無法滿足高效率和高品質的要求。

正是為了解決這些問題，asyncio 應運而生。

同步 VS 非同步

我們先區分Sync（同步）和Async（非同步）的概念。

同步是指操作依序執行。前一個操作完成後才能執行下一個操作。
非同步意味著不同的操作可以交替執行。如果其中一項操作被阻塞，程式不會等待，而是會尋找可執行的操作來繼續。

非同步如何運作

協程：asyncio 使用協程來實現非同步操作。協程是使用 async 關鍵字定義的特殊函數。在協程中，await 關鍵字可用於暫停目前協程的執行並等待非同步操作完成。
事件循環：事件循環是asyncio的核心機制之一。它負責調度和執行協程並處理協程之間的切換。事件循環將不斷輪詢可執行任務。一旦任務準備就緒（例如 I/O 操作完成或計時器到期），事件循環會將其放入執行佇列並繼續下一個任務。
非同步任務：在asyncio中，我們透過建立非同步任務來執行協程。非同步任務由 asyncio.create_task() 函數創建，該函數將協程封裝成一個可等待的對象，並將其提交給事件循環進行處理。
非同步I/O操作：asyncio提供了一組非同步I/O操作（如網路請求、檔案讀寫等），可以透過等待關鍵字。透過使用非同步I/O操作，可以避免等待I/O完成期間的阻塞，提高程式效能和並發性。
回呼：asyncio也支援使用回呼函數來處理非同步操作的結果。 asyncio.ensure_future() 函數可用於將回呼函數封裝成可等待對象，並將其提交給事件循環進行處理。
並發執行：asyncio可以並發執行多個協程任務。事件循環會根據任務的就緒情況自動調度協程的執行，從而實現高效的並發程式設計。

綜上所述，asyncio的工作原理是基於協程和事件循環的機制。 asyncio 透過使用協程進行非同步操作，並讓事件循環負責協程的調度和執行，實現了高效的非同步程式設計模型。

協程和異步編程

協程是 asyncio 中的重要概念。它們是輕量級執行單元，可以在任務之間快速切換，而無需執行緒切換的開銷。可以使用 async 關鍵字定義協程，await 關鍵字用於暫停協程的執行，並在某個操作完成後恢復。

這是一個簡單的範例程式碼，示範如何使用協程進行非同步程式設計：

import asyncio

async def hello():
    print("Hello")
    await asyncio.sleep(1)  # Simulate a time-consuming operation
    print("World")

# Create an event loop
loop = asyncio.get_event_loop()

# Add the coroutine to the event loop and execute
loop.run_until_complete(hello())

登入後複製

在此範例中，函數 hello() 是使用 async 關鍵字定義的協程。在協程內部，我們可以使用await來暫停其執行。這裡使用asyncio.sleep(1)來模擬一個耗時的操作。 run_until_complete() 方法將協程新增至事件循環並執行它。

非同步 I/O 操作

asyncio主要用於處理I/O密集型任務，如網路請求、檔案讀寫等。它提供了一系列非同步I/O操作的API，可以與await關鍵字結合使用，輕鬆實現非同步程式設計。

這是一個簡單的範例程式碼，展示如何使用 asyncio 進行非同步網路請求：

import asyncio
import aiohttp

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    async with aiohttp.ClientSession() as session:
        html = await fetch(session, 'https://www.example.com')
        print(html)

# Create an event loop
loop = asyncio.get_event_loop()

# Add the coroutine to the event loop and execute
loop.run_until_complete(main())

登入後複製

在此範例中，我們使用 aiohttp 函式庫來處理網路請求。函數 fetch() 是一個協程。它透過session.get()方法發起非同步GET請求，並使用await關鍵字等待回應傳回。函數 main() 是另一個協程。它在內部建立一個 ClientSession 物件以供複用，然後呼叫 fetch() 方法以取得網頁內容並列印。

注意：這裡我們使用aiohttp而不是requests函式庫，因為requests函式庫不相容asyncio，而aiohttp函式庫則相容。想要用好asyncio，尤其是發揮其強大的功能，很多時候需要對應的Python函式庫。

多個任務並發執行

asyncio也提供了一些並發執行多個任務的機制，例如asyncio.gather()和asyncio.wait()。以下是一個範例程式碼，展示如何使用這些機制來並發執行多個協程任務：

import asyncio

async def task1():
    print("Task 1 started")
    await asyncio.sleep(1)
    print("Task 1 finished")

async def task2():
    print("Task 2 started")
    await asyncio.sleep(2)
    print("Task 2 finished")

async def main():
    await asyncio.gather(task1(), task2())

# Create an event loop
loop = asyncio.get_event_loop()

# Add the coroutine to the event loop and execute
loop.run_until_complete(main())

登入後複製

在這個範例中，我們定義了兩個協程任務task1()和task2()，它們都執行一些耗時的操作。協程 main() 透過 asyncio.gather() 同時啟動這兩個任務並等待它們完成。並發執行可以提高程式執行效率。

如何選擇？

實際項目中，我們應該選擇多執行緒還是asyncio？有大佬總結的很形象：

import asyncio

async def hello():
    print("Hello")
    await asyncio.sleep(1)  # Simulate a time-consuming operation
    print("World")

# Create an event loop
loop = asyncio.get_event_loop()

# Add the coroutine to the event loop and execute
loop.run_until_complete(hello())

登入後複製

如果是 I/O 綁定，且 I/O 操作很慢，需要很多任務/執行緒的配合，那麼使用 asyncio 會比較合適。
如果是 I/O 綁定，但 I/O 操作很快並且只需要有限數量的任務/線程，那麼多線程就可以了。
如果是CPU密集型，那就需要多處理來提高程式運作效率。

實踐

輸入一個列表。對於清單中的每個元素，我們要計算從 0 到該元素的所有整數的平方和。

同步實施

import asyncio
import aiohttp

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    async with aiohttp.ClientSession() as session:
        html = await fetch(session, 'https://www.example.com')
        print(html)

# Create an event loop
loop = asyncio.get_event_loop()

# Add the coroutine to the event loop and execute
loop.run_until_complete(main())

登入後複製

執行時間為計算耗時16.00943413000002秒

使用並發.futures 進行非同步實現

import asyncio

async def task1():
    print("Task 1 started")
    await asyncio.sleep(1)
    print("Task 1 finished")

async def task2():
    print("Task 2 started")
    await asyncio.sleep(2)
    print("Task 2 finished")

async def main():
    await asyncio.gather(task1(), task2())

# Create an event loop
loop = asyncio.get_event_loop()

# Add the coroutine to the event loop and execute
loop.run_until_complete(main())

登入後複製

執行時間為計算耗時 7.314132894999999 秒

在這段改進後的程式碼中，我們使用concurrent.futures.ProcessPoolExecutor建立進程池，然後使用executor.map()方法提交任務並取得結果。請注意，使用 executor.map() 後，如果需要取得結果，可以將結果迭代到清單中，或使用其他方法處理結果。

多處理實現

if io_bound:
    if io_slow:
        print('Use Asyncio')
    else:
        print('Use multi-threading')
elif cpu_bound:
    print('Use multi-processing')

登入後複製

執行時間為計算耗時5.024221667秒

concurrent.futures.ProcessPoolExecutor和multiprocessing都是Python中用來實現多進程並發的函式庫。有些差異：

基於介面的封裝：concurrent.futures.ProcessPoolExecutor是concurrent.futures模組提供的高階介面。它封裝了底層的多進程函數，使編寫多進程程式碼變得更加容易。而multiprocessing是Python的標準函式庫之一，提供完整的多進程支持，並允許直接對進程進行操作。
API 用法：concurrent.futures.ProcessPoolExecutor 的用法與執行緒池類似。它將可呼叫的物件（例如函數）提交到進程池執行，並傳回一個Future對象，可以用來取得執行結果。多處理提供了更多低階進程管理和通訊介面。可以明確地建立、啟動和控制進程，並且可以使用佇列或管道來完成多個進程之間的通訊。
可擴展性和靈活性：由於多處理提供了更多底層接口，因此與concurrent.futures.ProcessPoolExecutor相比更加靈活。透過直接操作進程，可以對每個進程實現更細緻的控制，如設定進程優先權、進程間共享資料等。 concurrent.futures.ProcessPoolExecutor 更適合簡單的任務並行化，隱藏了許多底層細節，更容易編寫多進程程式碼。
跨平台支援：concurrent.futures.ProcessPoolExecutor 和 multiprocessing 都提供跨平台多進程支援，可以在各種作業系統上使用。

綜上所述，concurrent.futures.ProcessPoolExecutor是一個高層接口，封裝了底層多進程功能，適合簡單的多進程任務並行化。 multiprocessing是一個更底層的函式庫，提供了更多的控制和靈活性，適合需要對流程進行細粒度控制的場景。您需要根據具體需求選擇合適的庫。如果只是簡單的任務並行化，可以使用concurrent.futures.ProcessPoolExecutor來簡化程式碼；如果需要更多底層控制和通信，您可以使用多處理函式庫。