在 Python 的非同步程式設計中，asyncio 函式庫提供了協程、任務和事件迴圈等機制，讓開發者能有效地管理並執行非同步操作。理解協程與任務的轉換、事件迴圈的運作，以及如何妥善處理非同步任務中的錯誤，對於建構高效能且穩定的應用程式至關重要。本文將探討這些核心概念，並提供實務上的程式碼範例，幫助讀者掌握非同步程式設計的精髓。此外，文章也會探討非同步任務的動態控制、錯誤處理及任務同步等進階技巧，讓讀者能更有效地運用 asyncio 函式庫。
非同步程式設計：協程與任務的管理
在 Python 的 asyncio 框架中，非同步程式設計的核心在於有效地管理和執行協程（coroutine）。本章節將探討協程與任務（task）的運作機制，以及如何透過事件迴圈（event loop）來高效地執行非同步操作。
協程轉換為任務
要執行一個協程，首先需要將其轉換為任務。這可以透過 loop.create_task() 或 asyncio.create_task() 函式來實作。這種明確的封裝方式提供了對排程、取消和錯誤處理的直接控制。當一個協程被包裝成任務後，其執行狀態將由事件迴圈管理，確保在等待的操作就緒後繼續執行。
import asyncio
async def async_task(val):
await asyncio.sleep(0.2)
return f"已處理 {val}"
async def orchestrator():
task1 = asyncio.create_task(async_task(1))
task2 = asyncio.create_task(async_task(2))
task3 = asyncio.create_task(async_task(3))
results = await asyncio.gather(task1, task2, task3)
print("協調結果：", results)
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_until_complete(orchestrator())
loop.close()
內容解密：
async_task函式：定義了一個非同步任務，模擬延遲操作並傳回處理結果。orchestrator函式：建立了三個任務並發執行，並使用asyncio.gather等待所有任務完成。asyncio.create_task：將協程轉換為任務，使其能夠並發執行。loop.run_until_complete：執行事件迴圈直到指定的協程完成。
任務管理與事件迴圈
事件迴圈是 asyncio 框架的核心，負責管理和排程任務的執行。當一個任務被建立後，事件迴圈將負責其生命週期的管理，包括排程、暫停和取消。
協程與生成器的比較
在探討任務管理之前，瞭解協程與生成器（generator）的區別是非常重要的。雖然兩者都涉及可暫停和還原的函式呼叫，但它們的設計目標和使用場景有著本質的不同。
- 協程主要用於非同步操作，管理並發性和I/O操作。它們透過
async和await語法提供明確的暫停點，並與事件迴圈緊密整合。 - 生成器則主要用於迭代和流式資料處理。它們透過
yield提供隱式的暫停點，並不直接涉及非同步操作。
為何選擇原生協程
原生協程（native coroutine）提供了比生成器更豐富的功能和更好的整合性，特別是在錯誤處理、任務取消和排程方面。因此，在需要強大的非同步I/O操作的環境中，建議使用原生協程。
非同步任務管理的進階技巧與實務應用
在現代軟體開發中，非同步任務管理是提升系統效能和資源利用率的關鍵技術。透過asyncio函式庫，開發者可以有效地實作平行工作流程，取代傳統的同步呼叫模式。本文將探討非同步任務管理的進階技巧，包括動態平行控制、錯誤處理、任務取消以及任務同步等。
平行動態控制：使用訊號量限制平行任務數量
在處理大量非同步任務時，若不加以控制，可能會導致資源耗盡，特別是在I/O密集或高延遲的系統中。為瞭解決這個問題，可以使用asyncio.Semaphore來限制平行執行的任務數量。
import asyncio
async def limited_work(i, semaphore):
async with semaphore:
await asyncio.sleep(0.2)
print(f"Task {i} complete")
return i
async def run_with_limit():
semaphore = asyncio.Semaphore(5) # 最多允許5個任務平行執行
tasks = [asyncio.create_task(limited_work(i, semaphore)) for i in range(20)]
results = await asyncio.gather(*tasks)
print("Limited Task Results:", results)
asyncio.run(run_with_limit())
內容解密：
asyncio.Semaphore(5)：建立一個訊號量，初始值為5，表示最多允許5個任務同時執行。async with semaphore：任務執行前需先取得訊號量，若訊號量已滿，則任務會被暫停，直到有其他任務釋放訊號量。await asyncio.sleep(0.2)：模擬I/O操作或耗時任務。return i：傳回任務的結果。
錯誤處理：捕捉並處理非同步任務中的例外
在平行執行多個任務時，若其中一個任務丟擲例外，預設情況下會導致其他任務被取消。為了避免這種情況，可以使用return_exceptions=True引數來收集例外作為結果的一部分。
import asyncio
async def faulty_task(i):
await asyncio.sleep(0.1)
if i % 3 == 0:
raise ValueError(f"Fault in task {i}")
return f"Task {i} succeeded"
async def run_faulty_tasks():
tasks = [asyncio.create_task(faulty_task(i)) for i in range(10)]
results = await asyncio.gather(*tasks, return_exceptions=True)
for i, result in enumerate(results):
if isinstance(result, Exception):
print(f"Task {i} failed with exception: {result}")
else:
print(f"Task {i} completed with result: {result}")
asyncio.run(run_faulty_tasks())
內容解密：
return_exceptions=True：將例外作為結果傳回，避免因單一任務失敗而取消其他任務。isinstance(result, Exception)：檢查結果是否為例外，若是則列印錯誤訊息。faulty_task(i)：模擬可能失敗的任務，當i是3的倍數時丟擲ValueError。
任務取消：實作優雅的取消機制
在某些情況下，任務可能會因為需求變更或逾時而變得過時。此時，可以透過取消任務來釋放資源。asyncio提供了取消任務的機制，並允許在任務中捕捉CancelledError來進行清理操作。
import asyncio
async def cancellable_task(i):
try:
while True:
print(f"Task {i} working...")
await asyncio.sleep(0.3)
except asyncio.CancelledError:
print(f"Task {i} received cancellation signal")
# 在此執行必要的清理操作
raise
async def main():
tasks = [asyncio.create_task(cancellable_task(i)) for i in range(3)]
await asyncio.sleep(1)
for task in tasks:
task.cancel()
await asyncio.gather(*tasks, return_exceptions=True)
print("All tasks cancelled.")
asyncio.run(main())
內容解密：
task.cancel()：取消指定的任務。asyncio.CancelledError：當任務被取消時，會在任務內部丟擲此例外，允許進行清理操作。await asyncio.gather(*tasks, return_exceptions=True)：等待所有任務完成取消流程。
任務同步與依賴管理
當多個任務之間存在依賴關係時，需要使用同步機制來協調它們的執行順序。asyncio提供了多種同步工具，如asyncio.Event、asyncio.Condition和asyncio.Queue等。
import asyncio
async def producer(queue):
for i in range(5):
await queue.put(i)
print(f"Produced {i}")
await asyncio.sleep(0.1)
async def consumer(queue):
while True:
item = await queue.get()
print(f"Consumed {item}")
queue.task_done()
async def main():
queue = asyncio.Queue()
producer_task = asyncio.create_task(producer(queue))
consumer_task = asyncio.create_task(consumer(queue))
await producer_task
await queue.join() # 等待所有專案被處理完畢
consumer_task.cancel()
await asyncio.gather(consumer_task, return_exceptions=True)
asyncio.run(main())
內容解密：
asyncio.Queue()：建立一個佇列，用於生產者-消費者模式中的資料傳遞。await queue.put(i)：生產者將資料放入佇列。await queue.get()：消費者從佇列中取得資料。queue.task_done()：消費者處理完一個專案後呼叫此方法，表示該專案已完成處理。await queue.join()：等待佇列中所有專案都被處理完畢。
深入理解非同步程式設計中的錯誤處理
在非同步程式設計中，錯誤處理是一項複雜的挑戰，需要精心設計以確保系統的穩健性和容錯性。協程中的例外和錯誤傳播方式與同步程式碼不同。深入瞭解例外如何在非同步呼叫鏈中傳播對於開發具有彈性的系統至關重要。進階程式設計師必須利用結構化例外處理、取消管理和自定義錯誤傳播等技術，以設計出容錯的非同步應用程式。
協程中的錯誤傳播機制
在協程中，錯誤傳播是透過 await 表示式實作的。當一個協程等待另一個協程或未來（Future）時，任何在被等待的呼叫中發生的例外都會傳播回等待的協程。這種傳播可以使用標準的 try/except 區塊來攔截。然而，進階模式需要了解例外情境的細微差別，確保正確的清理和資源釋放。
程式碼範例：網路呼叫中的錯誤處理
import asyncio
async def unreliable_network_call():
await asyncio.sleep(0.1)
raise IOError("網路連線失敗")
async def network_handler():
try:
await unreliable_network_call()
except IOError as e:
print("已處理網路錯誤：", e)
# 可選地啟動重試機制或後備操作
return "後備結果"
return "成功結果"
async def main():
result = await network_handler()
print("結果：", result)
asyncio.run(main())
內容解密：
unreliable_network_call函式：模擬一個可能失敗的網路呼叫，透過asyncio.sleep(0.1)模擬延遲，並丟擲IOError例外。network_handler函式：捕捉unreliable_network_call中拋出的IOError例外，並進行處理。如果發生錯誤，則傳回「後備結果」；否則，傳回「成功結果」。main函式：呼叫network_handler並列印結果。
多層協程中的錯誤傳播
在多層協程中，錯誤可能發生在多層巢狀呼叫中。在這種情況下，以可控的方式傳播例外至關重要，以便更高層級的應用程式元件可以採用復原策略或記錄診斷資訊。此外，級聯例外可能需要保留原始的回溯資訊，以便精確識別故障點。
程式碼範例：多層協程中的錯誤處理
import asyncio
async def inner_coroutine():
await asyncio.sleep(0.1)
raise ValueError("內部錯誤")
async def middle_coroutine():
try:
await inner_coroutine()
except ValueError as e:
print("在中間層捕捉到錯誤：", e)
raise # 將錯誤進一步傳播
async def outer_coroutine():
try:
await middle_coroutine()
except ValueError as e:
print("在外部層捕捉到錯誤：", e)
return "復原結果"
return "正常結果"
async def main():
result = await outer_coroutine()
print("最終結果：", result)
asyncio.run(main())
內容解密：
inner_coroutine函式：模擬一個丟擲ValueError例外的內部協程。middle_coroutine函式：捕捉inner_coroutine中的例外，並進一步將其傳播。outer_coroutine函式：捕捉middle_coroutine中的例外，並傳回「復原結果」。main函式：呼叫outer_coroutine並列印最終結果。
自定義錯誤傳播與處理策略
進階程式設計師可以透過自定義錯誤處理策略來增強非同步應用程式的彈性。例如，使用集中式錯誤處理機制、記錄日誌或根據具體業務需求實施自定義的重試邏輯。
程式碼範例：自定義錯誤處理策略
import asyncio
import logging
logging.basicConfig(level=logging.INFO)
async def task_with_error():
await asyncio.sleep(0.1)
raise RuntimeError("任務執行失敗")
async def main():
try:
await task_with_error()
except RuntimeError as e:
logging.info(f"捕捉到錯誤並進行自定義處理：{e}")
# 在此處新增自定義處理邏輯，例如重試機制或通知機制
asyncio.run(main())
內容解密：
task_with_error函式：模擬一個丟擲RuntimeError例外的任務。main函式：捕捉例外並使用日誌記錄進行自定義處理，同時預留位置以新增進一步的自定義邏輯。