Python 的非同步程式設計是建立高效能應用程式的遊戲規則改變者。我已經使用它很多年了，如果使用得當，它的強大功能總是讓我驚嘆不已。
Python 非同步模型的核心是協程和事件循環。協程是可以暫停和恢復執行的特殊函數，可以在沒有執行緒開銷的情況下實現高效的多工處理。另一方面，事件循環是驅動這些協程、管理其執行和處理 I/O 操作的引擎。
讓我們從協程開始。在 Python 中，我們使用 async def 語法來定義它們。這是一個簡單的例子：
async def greet(name):
print(f"Hello, {name}!")
await asyncio.sleep(1)
print(f"Goodbye, {name}!")
這個協程向一個人打招呼，等待一秒鐘，然後說再見。 wait 關鍵字在這裡至關重要 - 它允許協程暫停其執行並將控制權交還給事件循環。
但是協程在幕後是如何運作的呢？它們實際上是建立在 Python 的生成器功能之上的。當您呼叫協程時，它不會立即運行。相反，它會傳回一個協程物件。這個物件可以發送值並且可以產生值，就像生成器一樣。
事件循環負責實際運行這些協程。它維護一個任務隊列（它們是協程的包裝器）並逐一執行它們。當協程遇到await語句時，事件循環將掛起它並繼續執行下一個任務。這就是協作式多工處理的本質──任務自願放棄控制權，讓其他任務運作。
這是事件循環如何運作的簡化版本：
class EventLoop:
def __init__(self):
self.ready = deque()
self.sleeping = []
def call_soon(self, callback):
self.ready.append(callback)
def call_later(self, delay, callback):
deadline = time.time() + delay
heapq.heappush(self.sleeping, (deadline, callback))
def run_forever(self):
while True:
self.run_once()
def run_once(self):
now = time.time()
while self.sleeping and self.sleeping[0][0] <= now:
_, callback = heapq.heappop(self.sleeping)
self.ready.append(callback)
if self.ready:
callback = self.ready.popleft()
callback()
else:
time.sleep(0.1) # Avoid busy waiting
此事件循環管理兩種類型的任務：準備執行的任務（在就緒雙端佇列中）和正在休眠的任務（在休眠清單中）。 run_forever 方法會持續執行任務，直到沒有剩餘任務為止。
現在我們來談談任務調度。 Python 中的 asyncio 模組提供了具有高階調度功能的更複雜的事件循環。它可以處理 I/O 操作、運行子進程，甚至與其他事件循環整合。
以下是如何使用 asyncio 同時運行多個協程：
import asyncio
async def task1():
print("Task 1 starting")
await asyncio.sleep(2)
print("Task 1 finished")
async def task2():
print("Task 2 starting")
await asyncio.sleep(1)
print("Task 2 finished")
async def main():
await asyncio.gather(task1(), task2())
asyncio.run(main())
此腳本將啟動兩個任務，但任務 2 將在任務 1 之前完成，因為它休眠的時間較短。
非同步程式設計最強大的應用之一是網路操作。讓我們來看一個簡單的非同步 Web 伺服器：
import asyncio
async def handle_client(reader, writer):
data = await reader.read(100)
message = data.decode()
addr = writer.get_extra_info('peername')
print(f"Received {message!r} from {addr!r}")
response = f"Echo: {message}\n"
writer.write(response.encode())
await writer.drain()
print("Close the connection")
writer.close()
async def main():
server = await asyncio.start_server(
handle_client, '127.0.0.1', 8888)
addr = server.sockets[0].getsockname()
print(f'Serving on {addr}')
async with server:
await server.serve_forever()
asyncio.run(main())
此伺服器無需使用執行緒即可同時處理多個客戶端，效率很高。
但是非同步程式設計不只適用於伺服器。它對於客戶端也非常有用，特別是當您需要發出多個網路請求時。這是一個簡單的網頁抓取工具，可以同時取得多個頁面：
async def greet(name):
print(f"Hello, {name}!")
await asyncio.sleep(1)
print(f"Goodbye, {name}!")
此抓取工具可以同時取得多個頁面，與同步方法相比，顯著加快了處理速度。
現在，讓我們深入探討一些更高階的概念。 Python 非同步模型的一個有趣功能是您可以建立自己的事件循環。如果您需要將非同步程式碼與其他框架集成，或者想要針對特定用例進行最佳化，這會很有用。
這是一個簡單的自訂事件循環，可以運行同步和非同步回調：
class EventLoop:
def __init__(self):
self.ready = deque()
self.sleeping = []
def call_soon(self, callback):
self.ready.append(callback)
def call_later(self, delay, callback):
deadline = time.time() + delay
heapq.heappush(self.sleeping, (deadline, callback))
def run_forever(self):
while True:
self.run_once()
def run_once(self):
now = time.time()
while self.sleeping and self.sleeping[0][0] <= now:
_, callback = heapq.heappop(self.sleeping)
self.ready.append(callback)
if self.ready:
callback = self.ready.popleft()
callback()
else:
time.sleep(0.1) # Avoid busy waiting
這個自訂循環非常基礎，但它示範了核心原則。您可以擴展它來處理更複雜的場景，例如 I/O 操作或計時器。
偵錯非同步程式碼可能具有挑戰性，尤其是在處理複雜的應用程式時。我發現有用的技術是使用 asyncio 的調試模式。您可以像這樣啟用它：
import asyncio
async def task1():
print("Task 1 starting")
await asyncio.sleep(2)
print("Task 1 finished")
async def task2():
print("Task 2 starting")
await asyncio.sleep(1)
print("Task 2 finished")
async def main():
await asyncio.gather(task1(), task2())
asyncio.run(main())
這將提供更詳細的錯誤訊息和警告，例如從未完成的協程或運行時間過長的回調。
另一個有用的調試技術是使用 asyncio 的任務自省功能。例如，您可以獲得所有正在運行的任務的清單：
import asyncio
async def handle_client(reader, writer):
data = await reader.read(100)
message = data.decode()
addr = writer.get_extra_info('peername')
print(f"Received {message!r} from {addr!r}")
response = f"Echo: {message}\n"
writer.write(response.encode())
await writer.drain()
print("Close the connection")
writer.close()
async def main():
server = await asyncio.start_server(
handle_client, '127.0.0.1', 8888)
addr = server.sockets[0].getsockname()
print(f'Serving on {addr}')
async with server:
await server.serve_forever()
asyncio.run(main())
這可以幫助您了解程式在任何特定時刻正在做什麼。
在最佳化非同步程式碼時，一個關鍵原則是最大限度地減少同步操作所花費的時間。任何長時間運行的同步程式碼都會阻塞事件循環，從而阻止其他協程運行。如果您有 CPU 密集型任務，請考慮在單獨的執行緒或進程中執行它們。
另一種最佳化技術是當您有多個可以同時運行的協程時使用 asyncio.gather。這比一一等待更有效率：
import asyncio
import aiohttp
async def fetch_page(session, url):
async with session.get(url) as response:
return await response.text()
async def main():
urls = [
'http://example.com',
'http://example.org',
'http://example.net'
]
async with aiohttp.ClientSession() as session:
tasks = [fetch_page(session, url) for url in urls]
pages = await asyncio.gather(*tasks)
for url, page in zip(urls, pages):
print(f"Page from {url}: {len(page)} bytes")
asyncio.run(main())
最後，請記住，非同步程式設計並不總是最好的解決方案。對於需要大量等待的 I/O 密集型任務，它可以提供顯著的效能改進。但對於 CPU 密集型任務，傳統的多執行緒或多處理可能更合適。
總之，Python 的非同步程式設計模型基於協程和事件循環，提供了一種編寫高效、可擴展應用程式的強大方法。無論您是建立 Web 伺服器、網路用戶端或資料處理管道，請理解這些概念都可以幫助您充分利用 Python 的非同步功能。與任何強大的工具一樣，它需要練習和仔細思考才能有效使用，但結果確實令人印象深刻。
一定要看看我們的創作：
投資者中心 | 智能生活 | 時代與迴聲 | 令人費解的謎團 | 印度教 | 精英開發 | JS學校
科技無尾熊洞察 | 時代與迴響世界 | 投資人中央媒體 | 令人費解的謎團 | | 令人費解的謎團 | >科學與時代媒介 |
現代印度教以上是掌握 Python 非同步：使用協程和事件循環提升應用程式效能的詳細內容。更多資訊請關注PHP中文網其他相關文章！
