- Python 中的非同步程式設計允許多個 I/O 密集型任務並行執行，而不會相互阻塞。
async,await以及事件循環。 - 利用
asyncio,aiohttp非同步上下文管理器和非同步迭代實現了可擴展的網路和 API 密集型工作負載。 - 非同步處理在網路和檔案 I/O 方面表現出色，但對於 CPU 密集型任務，應輔以多進程或專用服務。
- 良好的實踐——避免阻塞呼叫、限制並發性以及處理每個任務的錯誤——是編寫可靠的非同步應用程式的關鍵。
Python 中的非同步程式設計已經從一個冷門話題發展成為建立現代響應式應用程式必備的核心技能之一。 如果你正在使用 Web API、微服務、即時儀錶板或任何類型的高負載輸入/輸出 (I/O) 操作，你可能遇到過這樣的問題：程式碼等待的時間比實際執行操作的時間還要長。而這正是非同步技術大顯身手的地方。
非同步程式碼可以避免程式在等待網路、磁碟或外部服務時處於空閒狀態，從而允許程式重疊等待時間並保持應用程式運行。 在本指南中，我們將深入探討 Python 中非同步的工作原理，它解決了哪些問題，何時真正有用，以及何時不宜使用非同步，並透過具體的範例進行講解。 async, await, asyncio 以及一些流行的非同步庫，例如 aiohttp.
Python中的非同步程式設計是什麼？
非同步程式設計的核心在於建構程式碼結構，使多個任務即使共享同一個作業系統線程，也能在不相互阻塞的情況下繼續進行。 在傳統的同步程式設計風格中，每個操作都必須完成才能開始下一個操作：呼叫 API、等待、解析回應，然後才能繼續下一個操作。而非同步程式設計可讓你觸發多個耗時較長的操作，並讓 Python 在其中一個操作等待期間自動切換執行。
Python 透過特殊的語法和圍繞事件循環構建的協作式調度器來實現這一模型。 解開這一切謎團的兩個關鍵字是： async 以及 await您可以使用以下方式將函數標記為非同步： async def你會在它們裡面停下來， await 每當遇到可以將控制權交還給事件循環的操作時。
An async def 該函數不會直接傳回值；它傳回一個協程對象，該對象表示一個可以調度和等待的計算。 當您使用 await 在該函數內部，Python 會暫停目前的協程，並允許其他待處理的任務運行，直到等待的操作（例如網路請求）完成，此時執行會立即恢復。 await.
這一點至關重要：非同步 Python 程式碼通常仍然是單線程的，但它是並發的，因為多個操作在重疊的時間視窗中推進。 當一個任務等待 I/O 操作時，另一個任務可以獲得 CPU 時間。這就是為什麼非同步程式設計非常適合 I/O 密集型工作負載，但並不能神奇地加速 CPU 密集型任務。
一個具體的類比：國際象棋表演賽和等待時間
Python 社群中用來解釋並發執行與順序執行的經典類比來自同時進行的國際象棋表演。 想像一下，一位國際象棋特級大師與 24 位業餘棋手對弈。她可以用兩種不同的方式進行比賽，分別對應同步和非同步策略。
在同步版本中，她與一位對手坐下來，從頭到尾完成這一局遊戲，然後再前往下一張桌子。 她每走一步棋耗時5秒，業餘棋手思考時間約55秒。一盤典型的棋局有30次交換（總共60步）。這意味著每盤棋持續(55 + 5) × 30 = 1800秒，約30分鐘。 24盤棋，整個賽事將持續12小時。
在非同步版本中，她會在房間裡走動，在每個棋盤上走一步，然後立即走到下一個棋盤，而當前的對手則會思考他們的應對策略。 在 24 個棋盤上進行一輪棋需要 24 × 5 = 120 秒，即 2 分鐘。經過 30 輪這樣的棋局，全部棋局大約需要 3600 秒，即 1 小時。
關鍵在於，她的原始比賽速度從未改變；改變的是她如何利用對手的等待時間。 非同步 Python 程式碼遵循相同的原則：它不會加快 I/O 速度，但它可以確保你在等待網路、磁碟或任何外部資源時，正在做一些有用的事情。
同步請求與非同步請求：API 的實際範例
Python 中 async 最常見的用例之一是處理外部 API，其中每個請求都可能很容易花費數百毫秒甚至更長。 舉例來說，假設你想使用 GitHub 的公共 API 來取得幾個 GitHub 帳戶的追蹤者數量。
最直接的同步方法是使用流行的阻塞式 HTTP 用戶端，例如 requests. 你會在一個循環中對每個用戶端點執行 GET 請求，讀取 JSON 有效負載，提取… followers 將該欄位的內容列印或儲存。這種方法簡單易讀，但缺點是：對於您處理的每個帳戶，程式都會執行請求，然後等待回應，之後才會開始處理下一個帳戶。
所以如果你檢查三個用戶，例如 api.github.com/users/python, api.github.com/users/google 以及 api.github.com/users/firebase程式碼會傳送第一個請求，阻塞直到 GitHub 回應，然後繼續處理第二個請求，依此類推。 如果用戶數量很少，這或許可以接受，但隨著用戶數量增長到成百上千，總處理時間就會急劇增加，因為你的應用程式大部分時間都處於空閒狀態，等待遠端伺服器。
為了加快速度，您可以切換到基於非同步實現的方案。 asyncio 以及類似支援異步的 HTTP 用戶端 aiohttp. 在這個模型中，你會啟動多個協程任務，這些任務幾乎同時發出 HTTP 請求。然後，事件循環會等待來自任何一個協程任務的回應，並在資料到達時恢復當前任務的執行，而不是等待一個請求完全完成後再開始下一個。
當你將這兩種方法進行基準測試時，非同步版本通常會以很大的優勢勝出，尤其是在使用者數量增加的情況下。 每次請求的時間沒有變化，但獲取所有結果的總時間會急劇下降，因為您是並發處理多個連接而不是串行處理。
核心概念：協程、事件循環、任務與 Future
現代非同步 Python 底層的核心是幾個關鍵的建置模組，這些模組主要由以下元件提供： asyncio 模塊。 理解這些概念將使生態系統的其餘部分不再那麼神秘，並有助於設計強大的非同步架構。
協程是一種特殊的函數，它可以暫停和恢復自身的執行。 在今天的語法中，你可以這樣定義一個： async def當你呼叫它時，你會得到一個協程對象，該物件需要等待或調度；它不會像普通函數那樣立即執行完畢。在內部，每當你使用 await 對於可等待的物件（另一個協程、任務、未來等），Python 會暫停該協程，直到等待的操作完成。
事件循環是協調器，它追蹤所有待處理的協程、I/O 操作和計時器，並決定在任何給定時間運行哪一段程式碼。 歷史上，你必須透過顯式方式取得和管理循環。 asyncio.get_event_loop()但在現代 Python 程式碼中，首選模式是讓 asyncio.run() 圍繞頂層非同步函數建立、運行並關閉循環，例如 main().
任務是對協程的封裝，它告訴事件循環安排協程執行。 你可以把它們看作是輕量級任務：循環可以在多個任務之間交錯執行，而無需啟動多個執行緒。你通常會使用以下方式來建立任務： asyncio.create_task() 或是透過呼叫諸如…之類的助手 asyncio.gather()內部管理一系列任務。
Futures 代表稍後才會提供的結果，類似 JavaScript 中的 Promise。 任務和 Future 都是可等待的對象：你可以 await 它們會被暫停，直到底層操作完成。這種統一的協議大大簡化了編排程式碼，因為非同步流程的建構最終歸結為按正確的順序等待正確的物件。
非同步語法實踐： async, await, async with 以及 async for
这 async 關鍵字不僅限於函數定義；它還擴展到上下文管理器和循環，以便更高級的模式可以參與到非同步世界中。 了解這種擴展語法有助於你圍繞網路連接、會話、串流和自訂協定編寫優雅的程式碼。
最常見的形式是 async def它定義了一個非同步函數（協程工廠）。 在這樣的函數內部，你會大量使用 await 每當你呼叫另一個協程或可等待操作時，例如 asyncio.sleep()非同步 HTTP 請求或非同步資料庫查詢。請注意，您不能使用 await 直接位於腳本的頂層；它必須存在於一個…之中 async def.
雖然你可能很想打電話 time.sleep() 在協程中設定延遲，會完全違背使用非同步的初衷。 time.sleep() 它會阻塞整個線程，包括事件循環，因此在此期間其他非同步任務都無法執行。您必須使用非阻塞版本。 asyncio.sleep()這樣，在計時器倒數期間，控制權就交還給了循環。
Python 也透過以下方式支援非同步上下文管理器： async with透過定義特殊方法來實現 __aenter__ 以及 __aexit__. 當處理需要乾淨利落地進行設定和拆卸操作的物件時，例如開啟網路會話或取得非同步資源，這種方法尤其有用。一個典型的例子是管理一個 aiohttp.ClientSession 或使用單一 HTTP 請求 async with 使用程式碼區塊而不是手動調用 close().
最後，非同步迭代透過以下方式公開： async for它依賴魔法方法 __aiter__ 以及 __anext__ 詳見 PEP 492。 非同步迭代器和非同步生成器允許你隨著時間的推移產生專案。 await 在迭代過程中，這非常適合透過網路或其他非同步來源逐漸到達的串流資料。
同時運行多個任務 asyncio
非同步程式設計的真正威力在於，它可以同時執行多個 I/O 密集型任務，而不是一個接一個地運行。 在 Python 的非同步生態系中，主要工具包括： asyncio.create_task() 以及 asyncio.gather()兩者都在事件循環中調度協程。
與 asyncio.gather()您可以一次啟動多個協程，並等待它們全部完成，然後以清單或元組的形式接收它們的結果。 這種情況在批次 HTTP 呼叫、資料庫查詢或任何重複的非同步操作中極為常見。其底層原理為： gather() 將每個協程包裝成一個任務，並確保它們都能被驅動完成。
如果你回到取得 GitHub 個人資料的例子，但使用以下方式重構它： aiohttp 以及 asyncio.gather()最終你會得到三次類似函數的呼叫。 fetch_user() 同時啟動。 每個任務都會發起 HTTP 請求，在等待資料期間放棄控制權，然後在收到回應後繼續解析回應。從使用者的角度來看，所有三個結果幾乎同時顯示。
但是，在某些情況下，你不希望一次發出成千上萬個任務，因為這可能會使你的機器不堪重負或觸及外部速率限制。 一種常見的做法是透過僅處理特定進程來限制並發數。 MAX_TASKS 一次執行多個操作，可以使用信號量、有界池或在非同步工作流程中手動批次邏輯。
並發運行多個任務時，另一個關鍵方面是如何處理錯誤；在實際應用中，讓單一失敗的請求導致整個批次崩潰是很少可接受的。 理想情況下，非同步編排應該能夠捕獲並管理每個任務的異常，例如記錄異常、選擇性地重試或返回部分結果，同時保持批次的其餘部分完好無損。
處理併發問題：優勢與不足
區分並發和並行這兩個概念很重要，因為非同步 Python 提供了前者，但不一定提供了後者。 並發是指多個任務在重疊的時間間隔內進行，而並行則意味著它們實際上在多個 CPU 核心上同時運行。
典型的非同步程式碼使用 asyncio 它不會創建多個作業系統線程；相反，它會根據每個線程在 I/O 操作上阻塞的時間，在單個線程中復用任務，類似於… Node.js 程式設計. 這就是為什麼它能夠很好地擴展到數千個連接：上下文切換成本很低，因為它是由事件循環而不是作業系統協作控制的。
這種設計存在一些挑戰，尤其是在協調和異常處理方面。 由於你的邏輯現在分散在多個交錯執行的協程中，因此在共享狀態、傳播錯誤和清理資源時必須更加謹慎。諸如遺忘之類的錯誤可能會導致一些問題。 await永遠不會被等待的任務，或在後台任務中默默吞噬的異常，可能很微妙，難以調試。
為了保持非同步程式碼庫的可維護性，你應該遵循良好的工程實踐：保持協程專注於單一職責，盡可能集中錯誤處理，並添加足夠的日誌記錄以了解運行時發生的情況。 即使在單線程非同步環境中，良好的工具和清晰的約定也能有效防止競態條件問題或資源洩漏。
非同步程式碼何時真正有用（以及何時沒用）
非同步程式設計對於 I/O 密集型工作負載非常有效，但它並不是解決所有效能問題的萬靈藥。 任何最佳化工作的第一步都應該是確定瓶頸是來自 I/O 還是來自 CPU 密集型運算。
如果你的應用程式大部分時間都在等待網路響應、讀寫檔案、查詢資料庫或透過套接字通信，那麼非同步幾乎肯定是一個不錯的選擇。 典型的例子包括與多個外部服務通訊的 Web API、同時從多個資料來源讀取和寫入資料的 ETL 管道，以及維護多個並發客戶端連接的微服務。
另一方面，如果你的工作負載主要由繁重的 CPU 操作組成，例如數值計算、影像處理或複雜的模擬，那麼僅靠非同步操作是無法加快速度的。 在這種情況下，全域解釋器鎖定 (GIL) 仍然會限制單一 Python 進程中可以並行運行的任務數量。通常情況下，使用多進程、原生擴充或利用專用後端可以獲得更好的結果。
在企業環境中，務實的策略是將這些技術整合起來：使用 asyncio 和支援非同步的雲端服務 SDK（AWS、Azure 等）來最大限度地減少延遲並最大限度地提高吞吐量，同時將 CPU 密集型工作委託給單獨的進程、工作進程或託管運算服務。 這樣可以充分發揮每種工具的優勢，而不是與語言運行時對抗。
編寫異步 Python 程式碼的最佳實踐
一旦你開始更廣泛地採用非同步編程，某些模式和習慣將幫助你避免最常見的陷阱。 它們也能讓你的程式碼對那些可能還不熟悉非同步生態系統的隊友來說更清晰易懂。
一條基本原則是避免在非同步程式碼路徑中使用阻塞呼叫。 這意味著要更換諸如…之類的東西。 time.sleep() - await asyncio.sleep()此外，也要謹慎使用那些不提供非同步相容 API 的程式庫。如果第三方套件是純同步的，那麼在協程中大量呼叫它可能會阻塞事件循環，從而破壞並發優勢。
當您有一批獨立的 I/O 操作需要執行時，最好使用諸如 `mvn` 之類的工具並發運行它們。 asyncio.gather() 或受最大並發等級限制的任務池。 這種模式可以提高吞吐量，同時也能控制開啟的連線數或正在進行的請求數。
作為設計準則，盡量保持協程相對較小，並專注於明確的職責，類似於在簡潔的同步程式碼中設計函數的方式。 混合了網路、業務邏輯和錯誤處理的大型單體協程很快就會變得難以測試和推理，尤其是在過程中出現故障時。
最後，務必檢查你所依賴的生態系統組件是否真正支援非同步使用。 許多流行的庫都提供了獨立的非同步客戶端或專用子模組；而有些庫即使宣稱支援「非同步」功能，底層可能仍然存在阻塞。仔細閱讀文件並進行一些小規模的效能測試可以避免一些不易察覺的效能下降問題。
實際應用場景和架構理念
在現實世界的軟體專案中，非同步程式設計在各種架構中都表現出色，從傳統的 Web 後端到尖端的 AI 驅動系統。 共同點在於，需要在不浪費時間等待空閒時間的情況下處理許多 I/O 密集型操作。
一個經典的場景是，Web 服務需要呼叫多個外部 API 來建立一個傳送給客戶端的單一回應。 利用非同步技術，該服務可以一次觸發所有出站請求，並在每個資料區塊到達後立即組裝最終有效負載，從而顯著縮短總回應時間。這在微服務架構以及與支付網關、社交網路或分析平台的整合中十分常見。
另一個重要的用例是資料工程：管道和 ETL 作業經常並行地與多個資料庫、檔案系統或雲端儲存桶進行互動。 透過同時從多個來源讀取數據，並在結果準備就緒後立即寫入，可以降低整體延遲，更好地利用可用頻寬，尤其是在使用雲端儲存或基於 REST 的數據 API 時。
非同步操作也與商業智慧儀表板和 Power BI 等工具配合良好，後端必須聚合來自不同服務的數據，而不會阻塞長時間運行的 HTTP 連接。 使用以下方式建立自訂 API 層或整合微服務 asyncio 可以提高負載下的感知響應速度和吞吐量。
專門從事客製化軟體、人工智慧、網路安全和雲端諮詢的公司通常大量依賴非同步技術來協調呼叫 AI 模型、記錄事件、監控威脅和與雲端控制平面通訊的工作流程。 將非同步 I/O 用於編排，並與單獨的 CPU 最佳化工作進程用於繁重工作相結合，是一種常見的內部模式，可產生可擴展、可維護的系統。
對於許多開發人員和團隊來說，第一步就是將非同步引入應用程式中那些明顯受「I/O 限制」影響的部分，然後隨著非同步帶來的好處變得顯而易見，團隊對非同步範式和工具的信心增強，再逐步迭代。
歸根結底，Python 中的非同步程式設計是明智地利用等待時間：透過圍繞等待時間建立程式碼來實現。 async, await透過協程和事件循環，您可以建立速度更快、負載下可擴展性更好、能夠充分利用可用資源的應用程序，尤其是在處理網路、文件和外部服務時。
