掌握 Python 非同步：使用協程和事件循環提升應用程式效能

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中文網其他相關文章！