協(xié)程是實現(xiàn)高并發(fā)的一種非常高效的方式，特別適合處理大量I/O操作（如網(wǎng)絡(luò)請求、文件操作）的場景。它通過在單個線程內(nèi)實現(xiàn)多個任務(wù)的切換來避免阻塞，從而最大限度地利用CPU資源。

核心概念與簡單示例

要理解協(xié)程如何實現(xiàn)高并發(fā)，首先要明白幾個關(guān)鍵概念：

• 協(xié)程（Coroutine）：一種輕量級的“微線程”，可以在執(zhí)行過程中暫停（await）并在適當(dāng)?shù)臅r候恢復(fù)，而不是像普通函數(shù)那樣一次性執(zhí)行完畢。這種可掛起的能力是高并發(fā)的基石。
• 事件循環(huán)（Event Loop）：這是協(xié)程的“大腦”或調(diào)度中心。它負(fù)責(zé)在單個線程中調(diào)度和執(zhí)行多個協(xié)程任務(wù)。當(dāng)一個協(xié)程因等待I/O（比如網(wǎng)絡(luò)響應(yīng)）而暫停時，事件循環(huán)會立刻切換到另一個就緒的協(xié)程去執(zhí)行，從而避免了CPU的空等。
• async/await 關(guān)鍵字：async 用于聲明一個函數(shù)是異步函數(shù)（即協(xié)程），await 用于在協(xié)程內(nèi)部掛起自身，直到其后面的異步操作完成。

下面是一個簡單的例子，演示了如何使用 asyncio 庫創(chuàng)建兩個協(xié)程，讓它們“并發(fā)”地執(zhí)行：

import asyncio
import time

async def say_after(delay, what):
    await asyncio.sleep(delay)  # 模擬一個耗時的I/O操作
    print(what)

async def main():
    print(f"程序開始于: {time.strftime('%X')}")

    # 使用 asyncio.gather 來并發(fā)運行兩個協(xié)程
    await asyncio.gather(
        say_after(2, 'Hello,'),
        say_after(1, 'World!')
    )

    print(f"程序結(jié)束于: {time.strftime('%X')}")

# 運行主協(xié)程
asyncio.run(main())

在這個例子中，雖然 Hello, 任務(wù)設(shè)定等待2秒，而 World! 任務(wù)只等待1秒，但由于它們是并發(fā)執(zhí)行的，總耗時大約只有2秒，而不是順序執(zhí)行時的3秒。輸出將會是：

程序開始于: 12:00:00
World!
Hello,
程序結(jié)束于: 12:00:02

高并發(fā)實踐：網(wǎng)絡(luò)請求

協(xié)程的真正威力體現(xiàn)在處理成千上萬個網(wǎng)絡(luò)請求時。下面的例子展示了如何使用 aiohttp 庫異步地并發(fā)獲取多個網(wǎng)頁內(nèi)容。

import asyncio
import aiohttp
import time

async def fetch_url(session, url):
    async with session.get(url) as response:
        # 注意：在實際項目中，對于大響應(yīng)體，應(yīng)使用 response.content.read(chunk_size) 流式讀取
        html = await response.text()
        print(f"從 {url} 獲取了 {len(html)} 個字符")

async def main():
    urls = [
        'https://www.python.org',
        'https://www.example.com',
        'https://httpbin.org/get',
        # ... 可以添加更多URL
    ]

    # 關(guān)鍵：在整個抓取會話期間，復(fù)用同一個 ClientSession 實例
    # 其內(nèi)部維護了一個連接池，可以復(fù)用TCP連接，極大提升性能
    async with aiohttp.ClientSession() as session:
        tasks = []
        for url in urls:
            # 為每個URL創(chuàng)建一個協(xié)程任務(wù)
            task = asyncio.create_task(fetch_url(session, url))
            tasks.append(task)

        # 等待所有任務(wù)完成
        await asyncio.gather(*tasks)

# 記錄耗時
start_time = time.time()
asyncio.run(main())
end_time = time.time()
print(f"總共耗時: {end_time - start_time:.2f} 秒")

這個程序幾乎在同一時間發(fā)起所有請求，總耗時基本等于最慢的那個請求的耗時，而不是所有請求耗時的總和。

協(xié)程如何實現(xiàn)高并發(fā)：核心技術(shù)

協(xié)程實現(xiàn)高并發(fā)主要依賴于以下兩點：

協(xié)作式多任務(wù)與事件循環(huán)

協(xié)程是協(xié)作式的，這意味著它們會主動在可能發(fā)生阻塞的地方（即 await 處）掛起自己，將控制權(quán)交還給事件循環(huán)。事件循環(huán)則不斷地檢查：有哪些I/O操作已經(jīng)完成了？哪些協(xié)程可以恢復(fù)了？然后調(diào)度執(zhí)行那些就緒的協(xié)程。這個過程在單線程內(nèi)完成，完全避免了多線程帶來的鎖競爭和上下文切換的開銷。

非阻塞I/O與連接池

協(xié)程需要與非阻塞的I/O系統(tǒng)調(diào)用配合才能發(fā)揮最大效能。像 aiohttp 這樣的庫在底層使用了操作系統(tǒng)的非阻塞I/O機制（如Linux的 epoll）。當(dāng)一個協(xié)程發(fā)起網(wǎng)絡(luò)請求時，這個請求會以非阻塞方式立即發(fā)出，然后協(xié)程掛起。事件循環(huán)則通過I/O多路復(fù)用機制（如 select, poll, epoll）來監(jiān)聽大量socket（網(wǎng)絡(luò)連接）的狀態(tài)，一旦某個socket的數(shù)據(jù)準(zhǔn)備好了，事件循環(huán)就會喚醒等待該socket的協(xié)程繼續(xù)執(zhí)行。同時，如第一個搜索結(jié)果強調(diào)的，連接池（Connection Pool） 也至關(guān)重要。通過復(fù)用已經(jīng)建立的TCP連接，可以避免為每個請求都進行耗時的TCP三次握手，這在高并發(fā)下能帶來數(shù)量級的性能提升。

協(xié)程、多線程與多進程的對比

理解協(xié)程的適用場景很重要，下面的表格對比了三種主要的并發(fā)方式：

簡單來說：

I/O密集型（任務(wù)大部分時間在等待）：協(xié)程 >> 多線程 > 多進程

CPU密集型（任務(wù)大部分時間在計算）：多進程 > 多線程/協(xié)程（由于Python的GIL，計算密集型任務(wù)在多線程中無法真正并行）

實踐建議與常見誤區(qū)

避免在協(xié)程中調(diào)用阻塞性代碼：不要在協(xié)程內(nèi)使用 time.sleep(1) 或同步的 requests.get()。這會阻塞整個事件循環(huán)，使并發(fā)失效。務(wù)必使用對應(yīng)的異步版本 await asyncio.sleep(1) 和 aiohttp。

使用信號量（Semaphore）控制并發(fā)度：雖然協(xié)程很輕量，但向同一個目標(biāo)服務(wù)器發(fā)起無限度的并發(fā)請求可能會被視為攻擊或?qū)е聦Ψ椒?wù)不可用?？梢允褂?asyncio.Semaphore 來限制最大并發(fā)數(shù)。

復(fù)用對象：如網(wǎng)絡(luò)請求的例子所示，在整個應(yīng)用生命周期內(nèi)復(fù)用 ClientSession 這樣的對象，以利用連接池。

組合使用多種并發(fā)模型：對于更復(fù)雜的場景，可以采用“多進程+協(xié)程”的混合模式。例如，用多進程利用多核CPU，在每個進程內(nèi)部使用協(xié)程處理高并發(fā)的I/O。

到此這篇關(guān)于python協(xié)程實現(xiàn)高并發(fā)的技術(shù)詳解的文章就介紹到這了,更多相關(guān)python協(xié)程高并發(fā)內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評論