1. 前言 Python 以其簡潔的語法與強大的函式庫，深受眾多開發者喜愛。其中的「非同步處理」技術，是實現高效任務處理的重要方法之一。本文將從基礎到進階，清楚說明 Python 中的非同步處理機制。透過學習非同步處理，您將能大幅提升 Web 擷取與 API 請求的處理速度。
2. 非同步處理的基本知識 什麼是非同步處理？ 非同步處理是一種程式設計技巧，允許程式在等待一個任務完成的同時，繼續處理其他任務。舉例來說，在擷取多個網頁時，傳統的同步處理會依序發送每一個請求，而非同步處理則可以同時進行多個請求，大幅提高效率。同步處理與非同步處理的差異 特徵 同步處理 非同步處理 任務執行順序 依序執行一個任務 可同時執行多個任務 等待時間處理 會發生等待時間 可在等待時處理其他任務 適用範例 小型任務處理 需要大量 I/O 操作的情境
非同步處理的優點 提升效率 ：能同時處理多個任務，善用等待時間。擴展性佳 ：非常適合處理大量 I/O 操作的應用場景。節省資源 ：與建立多個執行緒或程序相比，非同步處理更能節省系統資源。3. 在 Python 中進行非同步處理的基本方法 在 Python 中實現非同步處理的方法 在 Python 中，要進行非同步處理，可以使用 async 與 await 這兩個關鍵字。透過這兩個語法，可以簡潔地撰寫非同步任務。import asyncio
async def say_hello():
print("こんにちは、非同期処理！")
await asyncio.sleep(1)
print("1秒経過しました！")
asyncio.run(say_hello())asyncawait協程（Coroutine）、任務（Task）、事件迴圈（Event Loop）的運作機制 協程 ：非同步任務的基本單位，使用 async 定義的函式就是協程。任務 ：將協程包裝後交由事件迴圈管理的單位。事件迴圈 ：Python 執行與排程非同步任務的核心機制。
4. 非同步處理的實際應用範例 在 Python 中，非同步處理可以應用於各種實務場景。以下將透過幾個實際的使用案例進行詳細說明：網頁擷取（Web Scraping） API 請求的並行處理 資料庫操作的非同步處理 網頁擷取（使用 aiohttp） 在進行網頁擷取時，通常需要對多個頁面發送請求並收集資料。使用非同步處理後，可以同時發送多個請求，大幅提升速度。 以下為使用 aiohttp 撰寫的非同步網頁擷取範例：import aiohttp
import asyncio
async def fetch_page(session, url):
async with session.get(url) as response:
print(f"Fetching: {url}")
return await response.text()
async def main():
urls = [
"https://example.com/page1",
"https://example.com/page2",
"https://example.com/page3"
]
async with aiohttp.ClientSession() as session:
tasks = [fetch_page(session, url) for url in urls]
results = await asyncio.gather(*tasks)
print("All pages fetched!")
asyncio.run(main())重點 ：使用 aiohttp.ClientSession 有效管理多個請求。 透過 asyncio.gather 同步執行多個非同步任務。 API 請求的並行處理 在處理多個 API 請求時，非同步處理同樣可以發揮極大的效益。以下為並行發送多個 API 請求並取得結果的範例：import aiohttp
import asyncio
async def fetch_data(session, endpoint):
async with session.get(endpoint) as response:
print(f"Requesting data from: {endpoint}")
return await response.json()
async def main():
api_endpoints = [
"https://api.example.com/data1",
"https://api.example.com/data2",
"https://api.example.com/data3"
]
async with aiohttp.ClientSession() as session:
tasks = [fetch_data(session, endpoint) for endpoint in api_endpoints]
results = await asyncio.gather(*tasks)
for i, result in enumerate(results):
print(f"Data from endpoint {i + 1}: {result}")
asyncio.run(main())重點 ：提升多個 API 請求的處理效率。 處理回傳的 JSON 格式資料。 資料庫操作的非同步處理（以 aiomysql 為例） 實作非同步的資料庫操作可以讓資料讀寫更為快速。以下為使用 aiomysql 進行非同步查詢的範例：import aiomysql
import asyncio
async def fetch_from_db():
conn = await aiomysql.connect(
host="localhost",
port=3306,
user="root",
password="password",
db="test_db"
)
async with conn.cursor() as cursor:
await cursor.execute("SELECT * FROM users")
result = await cursor.fetchall()
print("Data from database:", result)
conn.close()
asyncio.run(fetch_from_db())重點 ：執行非同步查詢以高效取得資料。 適用於需同時處理多筆資料庫操作的場景。 5. 使用非同步處理時的注意事項 非同步處理是一種非常強大的工具，但若使用不當，也可能導致一些意想不到的問題。以下將說明在使用非同步處理時需特別注意的要點，以及如何避免這些問題。避免死結（Deadlock） 死結是一種情況，當多個任務彼此等待資源而無法繼續執行時就會發生。在使用非同步處理時，必須妥善管理任務的順序與資源的取得時機。 範例：發生死結的情境 import asyncio
lock = asyncio.Lock()
async def task1():
async with lock:
print("Task1 acquired the lock")
await asyncio.sleep(1)
print("Task1 released the lock")
async def task2():
async with lock:
print("Task2 acquired the lock")
await asyncio.sleep(1)
print("Task2 released the lock")
async def main():
await asyncio.gather(task1(), task2())
asyncio.run(main())避免死結的方法 事先規劃各任務所需的資源，並以相同順序取得。 使用 asyncio.TimeoutError 設定資源取得的超時機制。 防止競爭條件（Race Condition） 在非同步處理中，若多個任務同時存取相同的資源，可能會發生資料不一致的「競爭條件」問題。 範例：競爭條件的情境 import asyncio
counter = 0
async def increment():
global counter
for _ in range(1000):
counter += 1
async def main():
await asyncio.gather(increment(), increment())
print(f"Final counter value: {counter}")
asyncio.run(main())counter 的值可能無法如預期正確累加。 防止競爭條件的方法 使用鎖（Lock） ：透過 asyncio.Lock 控制同時存取資源的任務。import asyncio
counter = 0
lock = asyncio.Lock()
async def increment():
global counter
async with lock:
for _ in range(1000):
counter += 1
async def main():
await asyncio.gather(increment(), increment())
print(f"Final counter value: {counter}")
asyncio.run(main())錯誤處理（Error Handling）的重要性 在非同步處理中，可能會遇到網路錯誤或超時等情況。若未妥善處理這些錯誤，可能會導致整體程式行為異常。 範例：實作錯誤處理 import asyncio
import aiohttp
async def fetch_url(session, url):
try:
async with session.get(url, timeout=5) as response:
return await response.text()
except asyncio.TimeoutError:
print(f"Timeout error while accessing {url}")
except aiohttp.ClientError as e:
print(f"HTTP error: {e}")
async def main():
urls = ["https://example.com", "https://invalid-url"]
async with aiohttp.ClientSession() as session:
tasks = [fetch_url(session, url) for url in urls]
await asyncio.gather(*tasks)
asyncio.run(main())錯誤處理的重點 針對可預期的錯誤進行對應處理。 透過例外處理紀錄錯誤日誌，以利後續除錯。 不適合使用非同步處理的情況 非同步處理並非適用於所有情境。特別是在以下幾種情況中，可能不適合使用：CPU 密集型任務 ：例如圖像處理或機器學習訓練等高 CPU 負載的作業，較適合使用 concurrent.futures 或 multiprocessing 模組。 小型任務 ：若任務本身耗時極短，初始化非同步處理所帶來的負擔反而超過其效益。 資源管理與最佳化 由於非同步處理會同時執行大量任務，因此有可能瞬間消耗大量記憶體與 CPU。以下是資源管理時的建議：限制同時執行的任務數 ： asyncio.Semaphore 可用來控制同時進行的任務數量。import asyncio
semaphore = asyncio.Semaphore(5)
async def limited_task(task_id):
async with semaphore:
print(f"Running task {task_id}")
await asyncio.sleep(1)
async def main():
tasks = [limited_task(i) for i in range(20)]
await asyncio.gather(*tasks)
asyncio.run(main())監控系統資源 ：
定期監測執行中的任務數與系統資源使用情況，以防超載。
6. 進階非同步處理的主題 當您掌握了非同步處理的基本概念後，進一步學習其應用與其他技術的比較，能幫助您更有效地活用非同步處理。本節將介紹 Python 以外的非同步技術，以及實際應用範例。與其他語言的非同步技術比較 不只 Python，許多程式語言也都支援非同步處理。以下將 Python 與幾種熱門技術進行比較，了解各自的特色：Node.js Node.js 是一種以非同步處理為強項的 JavaScript 執行環境，特別適合進行高效的 I/O 操作。特徵 Python Node.js 主要應用場景 資料分析、AI、Web 開發 Web 伺服器、即時應用 非同步處理方式 asyncio 模組、async/awaitCallback、Promise、async/await I/O 處理效能 效能不錯，但略低於 Node.js 針對非同步 I/O 最佳化 學習曲線 中等偏高 相對較低
Go Go（又稱 Golang）使用「Goroutine」這種輕量級執行緒來實現非同步處理。特徵 Python Go 主要應用場景 通用程式開發 伺服器、雲端開發 非同步處理方式 asyncio 模組、async/awaitGoroutine、Channel 平行處理效能 效能佳，但不適用於高 CPU 密集作業 在平行處理方面表現優異 學習曲線 中等 較低
Python 的優勢與應用範圍 通用性強 ：Python 不僅可用於 Web 開發，也廣泛應用於資料分析、機器學習等領域。函式庫豐富 ：透過 asyncio、aiohttp 等模組，可簡潔地實現複雜的非同步邏輯。非同步處理的應用情境 活用非同步處理，可以在以下情境中建構高效率的程式：伺服器端開發 非同步處理可應用於高負載的伺服器應用中。例如 FastAPI 是一款以非同步 I/O 為核心設計的 Python Web 框架，具有以下優點：高速 API 回應 ：支援高並行，能有效處理大量請求。簡潔的非同步程式碼 ：透過 async/await 編寫簡單明瞭的程式。from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def read_root():
return {"message": "Hello, FastAPI!"}微服務架構 在微服務架構中，應用會由多個小型服務協作構成。非同步處理在此有以下效益：提升服務間通訊效率 ：使用非同步 HTTP 請求或訊息佇列來降低延遲。增強系統擴展性 ：可獨立管理每個服務的資源。即時系統 在聊天應用或線上遊戲等即時系統中，非同步處理可實現流暢的資料更新。例如可使用 websockets 套件進行非同步 WebSocket 通訊。import asyncio
import websockets
async def echo(websocket, path):
async for message in websocket:
await websocket.send(f"Echo: {message}")
start_server = websockets.serve(echo, "localhost", 8765)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()學習非同步處理的進階步驟 若您希望進一步掌握非同步處理，以下主題值得深入學習：進階的非同步設計模式 ：實作任務取消、超時控制。 學習 asyncio 的低階 API（如 Future、自訂事件迴圈）。 函式庫的活用 ：如 aiohttp、aiomysql、asyncpg 等，適用於非同步 I/O。 非同步 Web 框架（如 FastAPI、Sanic）的實務應用。 結合分散式處理 ：
7. 總結 本文從基礎到應用，全面說明了 Python 的非同步處理技術。本章節將回顧重點內容，並提供進一步學習的方向。非同步處理概要 非同步處理可同時有效處理多項任務，特別適用於頻繁 I/O 操作的場景。其主要優勢如下：任務處理效率高 ：能善用等待時間處理其他任務。具擴展性 ：能高效處理大量並發請求。本文涵蓋的重點 非同步處理的基本知識 介紹同步與非同步的差異。 學習 async 和 await 的基本用法。 實際應用範例 以非同步方式執行 Web 擷取與 API 請求。 提升資料庫操作的效率。 注意事項與潛在問題 避免死結與競爭條件的設計方式。 妥善處理錯誤與管理資源。 進階應用方式 與其他非同步技術（Node.js、Go）比較。 應用於伺服器開發與即時應用中。 學習非同步處理的下一步 為進一步精通非同步處理，建議深入以下主題：函式庫活用 使用 aiohttp、aiomysql、asyncpg 進行實作。 使用非同步 Web 框架如 FastAPI 或 Sanic 開發 Web 應用。 進階設計模式 學習任務取消、例外處理與非同步佇列的使用。 活用 asyncio 的自訂事件迴圈進行低階控制。 實作專案應用 從小型非同步專案著手實驗。 解決實際問題，例如 API 加速或即時通訊等。
8. 常見問答（FAQ） 最後，我們整理了關於 Python 非同步處理的常見問題與解答：Q1：非同步處理與多執行緒有何不同？ 解答： 非同步處理在單一執行緒中透過切換任務來提高效率；而多執行緒則是透過多個執行緒同時處理任務。非同步適用於大量 I/O 任務，多執行緒則適合處理 CPU 密集型的任務。Q2：有哪些適合學習非同步處理的資源？ 解答： 推薦以下資源：Python 官方文件中的 asyncio 部分。 專門講解非同步處理的書籍（例如《Python Concurrency with Asyncio》）。 線上教學平台（如 Real Python 或 YouTube 教學影片）。 Q3：哪些情境適合使用非同步處理？ 解答： 以下情況非常適合使用非同步：需要處理大量 Web 請求（如 Web 擷取）。 即時通訊應用（如聊天室）。 頻繁等待外部 API 或資料庫的任務。 Q4：非同步處理是否適合用於 CPU 密集型任務？ 解答： 不適合。CPU 密集型任務應考慮使用 concurrent.futures 或 multiprocessing 模組來達到更好的效能。
