Python使用異步編程提升程序性能詳解

更新時間：2025年04月05日 09:52:48 作者：真智AI

這篇文章主要介紹了Python使用異步編程提升程序性能,異步編程是一種強大的技術(shù),可以顯著提升應用程序的性能,它允許你在不需要手動管理線程或進程的情況下,同時執(zhí)行多個任務,需要的朋友可以參考下

異步編程-提升應用性能的強大技術(shù)

異步編程是一種強大的技術(shù)，可以顯著提升應用程序的性能。它允許你在不需要手動管理線程或進程的情況下，同時執(zhí)行多個任務。這種方法特別適用于需要進行 API 請求、文件讀取或數(shù)據(jù)庫交互的開發(fā)者，而且不會暫停整個應用程序的運行。

在 Python 中，最知名的異步編程庫是 asyncio。這個庫通過實現(xiàn)事件循環(huán)（Event Loop）、協(xié)程（Coroutines）和非阻塞 I/O（Non-blocking I/O）來管理程序的執(zhí)行。如果你希望構(gòu)建高響應性和高可擴展性的應用程序，強烈建議你深入研究這個庫。

同步vs異步編程

在同步編程中，任務是按順序執(zhí)行的，意味著每個任務必須在前一個任務完成后才能開始。如下圖所示，程序會等待一個任務執(zhí)行完畢后再進行下一個任務。

（同步執(zhí)行示意圖）

相反，異步編程允許多個任務同時運行。當一個任務在等待某個慢操作（如從服務器獲取數(shù)據(jù)）時，程序可以繼續(xù)執(zhí)行其他任務。這種**“非阻塞”**方法避免了程序停滯，使可用資源得到更好的利用，從而顯著提高執(zhí)行效率。下圖展示了異步編程的工作方式。

（異步執(zhí)行示意圖）

什么是asyncio

asyncio 提供了一個框架，使開發(fā)者可以使用 async 和 await 語法編寫異步代碼。它允許程序在不停止其他進程的情況下執(zhí)行網(wǎng)絡請求、文件訪問等操作。這樣，程序可以在等待某些任務完成時仍然保持響應性。

值得注意的是，asyncio 從 Python 3.3 版本起就默認包含，無需額外安裝。

asyncio的核心組件

事件循環(huán)（Event Loop）：管理和調(diào)度異步任務和協(xié)程的核心機制。
協(xié)程（Coroutines）：使用 async def 定義的特殊函數(shù)，它們可以使用 await 關(guān)鍵字暫停執(zhí)行，以便讓其他任務運行。
任務（Tasks）：已被調(diào)度到事件循環(huán)中的協(xié)程，可使用 asyncio.create_task() 創(chuàng)建任務。
Future 對象（Futures）：表示可能尚未完成的操作的結(jié)果，相當于未來可用的占位符。

asyncio的基本用法

import asyncio
async def calculate_square(number):
    print(f"開始計算 {number} 的平方")
    await asyncio.sleep(1)  # 模擬延遲
    print(f"{number} 的平方計算完成")
    return number ** 2
async def calculate_cube(number):
    print(f"開始計算 {number} 的立方")
    await asyncio.sleep(2)  # 模擬延遲
    print(f"{number} 的立方計算完成")
    return number ** 3
async def main():
    # 創(chuàng)建任務并調(diào)度到事件循環(huán)中
    task1 = asyncio.create_task(calculate_square(5))  
    task2 = asyncio.create_task(calculate_cube(3))   
    # 等待任務完成
    result1 = await task1
    result2 = await task2
    # 輸出結(jié)果
    print(f"平方結(jié)果：{result1}")  
    print(f"立方結(jié)果：{result2}")   
if __name__ == "__main__":
    # 啟動事件循環(huán)
    asyncio.run(main())

關(guān)鍵點解析

事件循環(huán)：asyncio.run(main()) 啟動事件循環(huán)，并運行 main() 協(xié)程。
協(xié)程：calculate_square(number) 和 calculate_cube(number) 是協(xié)程，它們使用 await asyncio.sleep() 模擬延遲，而不會阻塞整個程序。
任務：在 main() 中，我們使用 asyncio.create_task() 創(chuàng)建任務，讓兩個協(xié)程并行運行。
Future 對象：result1 和 result2 是 Future 對象，代表 calculate_square() 和 calculate_cube() 任務的最終結(jié)果。

運行結(jié)果

開始計算 5 的平方
開始計算 3 的立方
5 的平方計算完成
3 的立方計算完成
平方結(jié)果：25
立方結(jié)果：27

運行多個協(xié)程asyncio.gather()

asyncio.gather() 允許多個協(xié)程并行執(zhí)行，通常用于批量發(fā)送網(wǎng)絡請求。讓我們用一個示例來模擬多個 HTTP 請求的并發(fā)執(zhí)行。

示例代碼

import asyncio
# 模擬從 URL 獲取數(shù)據(jù)的任務
async def fetch_data(URL, delay):
    print(f"開始從 {URL} 獲取數(shù)據(jù)")
    await asyncio.sleep(delay)  # 模擬網(wǎng)絡延遲
    return f"從 {URL} 獲取數(shù)據(jù)完成（耗時 {delay} 秒）"
async def main():
    # 并行執(zhí)行多個獲取數(shù)據(jù)的任務
    results = await asyncio.gather(
        fetch_data("https://openai.com", 2),  
        fetch_data("https://github.com", 5),  
        fetch_data("https://python.org", 7)   
    )
    # 輸出所有任務的結(jié)果
    for result in results:
        print(result)
asyncio.run(main())

關(guān)鍵點解析

fetch_data(URL, delay) 模擬從 URL 獲取數(shù)據(jù)，并設置不同的延遲時間。
asyncio.gather() 讓三個 fetch_data() 協(xié)程并行執(zhí)行，而不會一個完成后再啟動下一個。
任務并行運行，因此程序不會等待一個任務完成后才開始下一個，極大提升效率。

運行結(jié)果

開始從 https://openai.com 獲取數(shù)據(jù)
開始從 https://github.com 獲取數(shù)據(jù)
開始從 https://python.org 獲取數(shù)據(jù)
從 https://openai.com 獲取數(shù)據(jù)完成（耗時 2 秒）
從 https://github.com 獲取數(shù)據(jù)完成（耗時 5 秒）
從 https://python.org 獲取數(shù)據(jù)完成（耗時 7 秒）

處理超時asyncio.wait_for()

asyncio.wait_for() 允許你為異步任務設置超時時間，如果任務超時未完成，將引發(fā) asyncio.TimeoutError。

示例代碼

import asyncio
async def fetch_data():
    await asyncio.sleep(15)  # 模擬網(wǎng)絡延遲
    return "數(shù)據(jù)獲取成功"
async def main():
    try:
        # 設置超時時間為 10 秒
        result = await asyncio.wait_for(fetch_data(), timeout=10) 
        print(result)
    except asyncio.TimeoutError:
        print("數(shù)據(jù)獲取超時")
asyncio.run(main())

如果 fetch_data() 任務執(zhí)行超過 10 秒，就會觸發(fā)超時異常 TimeoutError。