高效能 Python：Asyncio

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

為什麼我們需要 asyncio

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

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

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

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

同步 VS 非同步

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

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

非同步如何運作

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

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

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

概括

與多執行緒不同，asyncio 是單執行緒的，但其內部事件循環的機制允許它同時運行多個不同的任務，並且比多執行緒具有更強的自主控制能力。

asyncio中的任務在運行過程中不會被中斷，因此不會出現race condition的情況。

尤其是在I/O操作繁重的場景下，asyncio比多執行緒有更高的運作效率。因為asyncio中任務切換的成本遠小於執行緒切換的成本，而且asyncio可以啟動的任務數量遠大於多執行緒中的執行緒數量。

但要注意的是，很多情況下，使用asyncio需要特定的第三方函式庫的支持，例如上例中的aiohttp。而如果I/O操作又快又不重的話，使用多執行緒也能有效解決問題。

• asyncio 是一個用於實作非同步程式設計的 Python 函式庫。
• 協程是asyncio的核心概念，透過async和await關鍵字實現非同步操作。
• asyncio 為非同步 I/O 操作提供了強大的 API，可以輕鬆處理 I/O 密集型任務。
• 透過asyncio.gather()等機制，可以並發執行多個協程任務。

---

Leapcell：FastAPI、Flask 和其他 Python 應用程式的理想平台

最後介紹一下部署Flask/FastAPI的理想平台：Leapcell。

Leapcell是專為現代分散式應用程式設計的雲端運算平台。其按需付費的定價模式確保沒有閒置成本，這意味著用戶只需為他們實際使用的資源付費。

1. 多語言支援
   • 支援 JavaScript、Python、Go 或 Rust 開發。
2. 免費部署無限專案
   • 僅依使用情況收費。沒有要求時不收費。
3. 無與倫比的成本效益
   • 即用即付，無閒置費用。
   • 例如，25 美元可以支援 694 萬個請求，平均回應時間為 60 毫秒。
4. 簡化的開發者體驗
   • 直覺的使用者介面，易於設定。
   • 完全自動化的 CI/CD 管道和 GitOps 整合。
   • 即時指標和日誌，提供可操作的見解。
5. 輕鬆的可擴充性和高效能
   • 自動伸縮，輕鬆應付高併發。
   • 零營運開銷，讓開發者專注於開發。

在文件中了解更多！
Leapcell Twitter：https://x.com/LeapcellHQ

以上是高效能 Python：Asyncio的詳細內容。更多資訊請關注PHP中文網其他相關文章！