Python異步編程入門：協(xié)程到底是什么？

你可能已經(jīng)遇到過(guò)這樣的場(chǎng)景：寫了一個(gè)爬蟲，但請(qǐng)求網(wǎng)頁(yè)時(shí)總是卡??；或者搭了個(gè)Web服務(wù)，同時(shí)處理幾個(gè)請(qǐng)求就變得慢吞吞。這時(shí)候別人可能會(huì)告訴你：“試試異步編程吧，用協(xié)程。”

但協(xié)程到底是什么？它跟線程、進(jìn)程有什么區(qū)別？為什么大家都在說(shuō)asyncio？今天咱們就來(lái)好好聊聊這個(gè)話題。

從瓶頸說(shuō)起：程序?yàn)槭裁纯ǎ?/h3>

想象一下你去快餐店點(diǎn)餐。如果只有一個(gè)收銀員，前面的人點(diǎn)單特別慢，你就得一直等著——這就是典型的同步阻塞。你的代碼可能正在等網(wǎng)絡(luò)響應(yīng)、等文件讀寫、等數(shù)據(jù)庫(kù)查詢，而CPU就在那里空轉(zhuǎn)。

換成多線程呢？好比開了幾個(gè)收銀臺(tái)，同時(shí)服務(wù)。但每個(gè)收銀臺(tái)還是要等顧客慢慢掏錢、找零，而且收銀臺(tái)之間還要協(xié)調(diào)（比如共用一臺(tái)打印機(jī)），協(xié)調(diào)不好就會(huì)出問(wèn)題。在程序里，這就是線程安全問(wèn)題和上下文切換開銷。

協(xié)程的做法不太一樣：一個(gè)收銀員同時(shí)接待多個(gè)顧客。A顧客在掏錢包時(shí)，收銀員轉(zhuǎn)頭問(wèn)B顧客要什么；B顧客猶豫時(shí)，又去給C顧客結(jié)賬?？雌饋?lái)收銀員一直在忙，但實(shí)際沒(méi)有真的“同時(shí)”做多件事，只是切換得很快。

協(xié)程的本質(zhì)：可暫停的函數(shù)

協(xié)程本質(zhì)上是一種特殊的函數(shù)，它能在執(zhí)行到一半時(shí)暫停，把控制權(quán)交出去，過(guò)一會(huì)兒又能從暫停的地方繼續(xù)執(zhí)行。

async def fetch_data():
    print("開始請(qǐng)求數(shù)據(jù)")
    await asyncio.sleep(2)  # 在這里暫停，讓其他協(xié)程運(yùn)行
    print("數(shù)據(jù)返回了")

這個(gè)await asyncio.sleep(2)就像告訴程序：“我先歇會(huì)兒，你去干點(diǎn)別的，2秒后再叫我。”這種機(jī)制讓單個(gè)線程可以處理多個(gè)任務(wù)，而不是傻等。

你可能會(huì)問(wèn)，這和生成器（generator）的yield有點(diǎn)像吧？確實(shí)，Python的協(xié)程就是從生成器進(jìn)化來(lái)的。早期的協(xié)程就是用yield實(shí)現(xiàn)的，但async/await語(yǔ)法更清晰，專門為異步編程設(shè)計(jì)。

協(xié)程 vs 線程 vs 進(jìn)程：什么時(shí)候選哪個(gè)？

先看個(gè)簡(jiǎn)單的對(duì)比：

• 多進(jìn)程：開多個(gè)廚房，各自獨(dú)立，但溝通成本高（進(jìn)程間通信）。適合CPU密集型計(jì)算，能利用多核優(yōu)勢(shì)。
• 多線程：一個(gè)廚房里多個(gè)廚師，共用設(shè)備，但要小心互相干擾（線程安全）。適合I/O操作，但線程數(shù)量多了開銷大。
• 協(xié)程：一個(gè)廚師同時(shí)照看幾口鍋，哪口鍋需要攪拌就攪一下，不需要時(shí)就處理別的。適合高并發(fā)I/O，但單個(gè)協(xié)程不能阻塞。

協(xié)程最大的優(yōu)勢(shì)就是輕量。創(chuàng)建一個(gè)線程需要幾MB內(nèi)存，而一個(gè)協(xié)程可能只要幾KB。一個(gè)線程能跑成千上萬(wàn)個(gè)協(xié)程，切換開銷極小，特別適合I/O密集的場(chǎng)景——比如網(wǎng)絡(luò)請(qǐng)求、文件讀寫，這些操作大部分時(shí)間都在等，而不是真的在計(jì)算。

但協(xié)程不是銀彈。如果你的任務(wù)是純計(jì)算型的（比如圖像處理、復(fù)雜算法），協(xié)程幫不上什么忙，因?yàn)橛?jì)算本身不會(huì)主動(dòng)讓出CPU。這時(shí)候多進(jìn)程或者直接優(yōu)化算法可能更有效。

Python協(xié)程的演進(jìn)：從yield到async/await

Python實(shí)現(xiàn)協(xié)程的方式有過(guò)幾次大的變化，了解這段歷史有助于理解為什么現(xiàn)在是這樣的設(shè)計(jì)：

# 1. 生成器時(shí)代（Python 2.5+） - 用yield模擬
def old_style_coroutine():
    print("開始")
    data = yield "請(qǐng)給我數(shù)據(jù)"
    print(f"收到數(shù)據(jù)：{data}")
# 使用方式：
coro = old_style_coroutine()
next(coro)  # 啟動(dòng)，輸出"開始"，返回"請(qǐng)給我數(shù)據(jù)"
coro.send("hello")  # 發(fā)送數(shù)據(jù)，輸出"收到數(shù)據(jù)：hello"
# 2. 裝飾器時(shí)代（Python 3.4）- 引入asyncio
@asyncio.coroutine
def decorator_style():
    yield from asyncio.sleep(1)
    print("完成")
# 3. 現(xiàn)代寫法（Python 3.5+）- 現(xiàn)在的標(biāo)準(zhǔn)
async def modern_coroutine():
    await asyncio.sleep(1)
    print("完成")

現(xiàn)在基本都用async/await這套語(yǔ)法，清晰直觀。但你可能在舊代碼里看到前兩種寫法，知道它們是一回事就行。有趣的是，async/await在C#、JavaScript等語(yǔ)言中也是類似的語(yǔ)法，學(xué)會(huì)一次，多語(yǔ)言受益。

深入理解async和await

第一次見(jiàn)async def可能會(huì)有點(diǎn)懵：這函數(shù)怎么調(diào)用后不執(zhí)行??？

async def hello():
    print("Hello, async!")
coro = hello()  # 注意：這里不會(huì)打印任何東西！
print(type(coro))  # <class 'coroutine'>

這里有個(gè)重要概念：協(xié)程函數(shù)被調(diào)用時(shí)返回的是一個(gè)協(xié)程對(duì)象，而不是直接執(zhí)行。要讓它跑起來(lái)，需要事件循環(huán)的調(diào)度。這就像給了你一張任務(wù)卡，但需要有人（事件循環(huán)）來(lái)執(zhí)行它。

await則是協(xié)程世界里的“等待”符號(hào)。但它不是阻塞等待，而是“我這兒暫時(shí)沒(méi)事，你先去忙別的”。

async def main():
    print("開始煮面")
    print("水燒開了，下面條")
    await asyncio.sleep(3)  # 等面煮熟，但程序可以去干別的
    print("面熟了，撈起來(lái)")
    print("開始炒菜")
    await asyncio.sleep(2)  # 等菜炒熟
    print("菜好了")
    print("開飯！")

有個(gè)常見(jiàn)的誤解：await就是異步。其實(shí)await本身不創(chuàng)造異步，它只是告訴程序“這里可以切換”。真正的異步能力來(lái)自那些支持異步的操作，比如asyncio.sleep()、aiohttp的網(wǎng)絡(luò)請(qǐng)求等。

另一個(gè)常見(jiàn)誤區(qū)：以為用了async/await就自動(dòng)變快。實(shí)際上，如果所有操作都是順序的await，那和同步?jīng)]什么區(qū)別。真正的并發(fā)要靠asyncio.gather()、asyncio.wait()這樣的結(jié)構(gòu)。

事件循環(huán)：協(xié)程的調(diào)度中心

如果協(xié)程是演員，事件循環(huán)就是導(dǎo)演。導(dǎo)演決定哪個(gè)演員什么時(shí)候上場(chǎng)、什么時(shí)候休息。

在Python 3.7之前，你需要自己管理事件循環(huán)：

# Python 3.6及以前
import asyncio
async def task():
    print("任務(wù)執(zhí)行中")
    await asyncio.sleep(1)
# 手動(dòng)管理事件循環(huán)
loop = asyncio.get_event_loop()
loop.run_until_complete(task())
loop.close()

Python 3.7引入了asyncio.run()，簡(jiǎn)化了這一切：

# Python 3.7+
async def main():
    await task()
asyncio.run(main())  # 一行搞定

asyncio.run()幫我們做了三件事：創(chuàng)建新的事件循環(huán)、運(yùn)行協(xié)程、關(guān)閉循環(huán)。對(duì)于大多數(shù)應(yīng)用，這就夠了。

事件循環(huán)的工作方式有點(diǎn)像餐廳的叫號(hào)系統(tǒng)：不斷檢查有沒(méi)有新的“事件”（比如網(wǎng)絡(luò)數(shù)據(jù)到達(dá)、定時(shí)器到期），然后喚醒對(duì)應(yīng)的協(xié)程繼續(xù)工作。它維護(hù)著一個(gè)待辦事項(xiàng)列表，哪個(gè)能處理就處理哪個(gè)。

實(shí)戰(zhàn)對(duì)比：同步 vs 異步下載

理論說(shuō)了這么多，來(lái)個(gè)實(shí)際例子感受一下區(qū)別。

假設(shè)你要下載10張網(wǎng)絡(luò)圖片，用傳統(tǒng)同步方式大概是這樣：

import requests
import time
def download_sync(url, filename):
    response = requests.get(url)
    with open(filename, 'wb') as f:
        f.write(response.content)
urls = [...]  # 10個(gè)圖片URL
start = time.time()
for i, url in enumerate(urls):
    download_sync(url, f"image_{i}.jpg")
    print(f"下載完第{i+1}張")
print(f"總耗時(shí): {time.time()-start:.2f}秒")
# 如果每張圖要1秒，這里大概要10秒

這是典型的順序執(zhí)行，一張下完再下一張，大部分時(shí)間都在等網(wǎng)絡(luò)響應(yīng)。

換成協(xié)程版本：

import aiohttp
import asyncio
import time
async def download_async(session, url, filename):
    async with session.get(url) as response:
        content = await response.read()
        with open(filename, 'wb') as f:
            f.write(content)
async def main():
    urls = [...]  # 同樣的10個(gè)URL
    async with aiohttp.ClientSession() as session:
        tasks = []
        for i, url in enumerate(urls):
            task = download_async(session, url, f"image_{i}.jpg")
            tasks.append(task)
        await asyncio.gather(*tasks)  # 并發(fā)下載！
start = time.time()
asyncio.run(main())
print(f"總耗時(shí): {time.time()-start:.2f}秒")
# 可能只要1-2秒就全部下載完了

區(qū)別很明顯：同步版本是等一張下完再下一張；協(xié)程版本是同時(shí)發(fā)起所有請(qǐng)求，哪張先到就先處理哪張。對(duì)于I/O密集型任務(wù)，速度提升可能非常顯著。

不過(guò)要注意，并不是所有場(chǎng)景都能這樣簡(jiǎn)單替換。requests庫(kù)是同步的，不能直接在協(xié)程里用，需要換用異步版本的aiohttp。這也是很多初學(xué)者容易踩的坑：用了async/await，但調(diào)用的庫(kù)不支持異步，結(jié)果還是同步執(zhí)行。

常見(jiàn)的協(xié)程使用場(chǎng)景

什么時(shí)候該考慮用協(xié)程呢？這里有幾個(gè)典型場(chǎng)景：

• Web服務(wù)器：比如用FastAPI、Sanic或aiohttp框架。每個(gè)請(qǐng)求都可能涉及數(shù)據(jù)庫(kù)查詢、外部API調(diào)用，用協(xié)程可以同時(shí)處理大量連接。
• 網(wǎng)絡(luò)爬蟲：需要爬取大量網(wǎng)頁(yè)，每個(gè)網(wǎng)頁(yè)的下載都是I/O操作，協(xié)程能顯著提高效率。
• 實(shí)時(shí)通信：聊天應(yīng)用、消息推送，需要維持大量長(zhǎng)連接，協(xié)程的內(nèi)存開銷比線程小得多。
• 批量文件處理：比如讀取大量文件、圖片處理（注意：圖片處理本身是CPU密集型，但讀取寫入文件是I/O密集型）。
• 微服務(wù)調(diào)用：一個(gè)服務(wù)需要調(diào)用多個(gè)其他微服務(wù)，然后合并結(jié)果，協(xié)程可以并行發(fā)起所有調(diào)用。

協(xié)程學(xué)習(xí)的難點(diǎn)和坑

剛開始學(xué)協(xié)程時(shí)，有幾個(gè)常見(jiàn)的困惑點(diǎn)：

第一，忘記加await：

async def get_data():
    return "數(shù)據(jù)"
async def main():
    result = get_data()  # 錯(cuò)誤！應(yīng)該加await
    print(result)  # 打印出來(lái)是個(gè)協(xié)程對(duì)象，不是字符串

第二，在同步函數(shù)里調(diào)用協(xié)程：

def sync_func():
    data = await get_data()  # 語(yǔ)法錯(cuò)誤！await只能在async函數(shù)里用

第三，阻塞事件循環(huán)：

async def bad_example():
    # 這個(gè)函數(shù)會(huì)阻塞整個(gè)事件循環(huán)！
    time.sleep(5)  # 應(yīng)該用await asyncio.sleep(5)
    # CPU密集型計(jì)算也會(huì)阻塞
    sum(range(10**7))  # 應(yīng)該放到線程池里執(zhí)行

第四，以為所有庫(kù)都支持異步：實(shí)際上很多常用庫(kù)（比如requests、pymysql的同步API）都是同步的。需要用異步替代庫(kù)（aiohttp、aiomysql）或者把同步調(diào)用放到線程池里。

性能真的提升了嗎？一個(gè)簡(jiǎn)單測(cè)試

我們來(lái)做個(gè)簡(jiǎn)單實(shí)驗(yàn)，看看協(xié)程在I/O密集型任務(wù)上的實(shí)際表現(xiàn)：

import asyncio
import time
import aiohttp
import requests
# 測(cè)試用的URL，請(qǐng)求會(huì)延遲1秒返回
TEST_URL = "http://httpbin.org/delay/1"
# 同步版本
def sync_test(n=10):
    start = time.time()
    for i in range(n):
        requests.get(TEST_URL)
    return time.time() - start
# 異步版本
async def async_test(n=10):
    start = time.time()
    async with aiohttp.ClientSession() as session:
        tasks = [session.get(TEST_URL) for _ in range(n)]
        await asyncio.gather(*tasks)
    return time.time() - start
# 運(yùn)行測(cè)試
print("同步版，10個(gè)請(qǐng)求：")
print(f"耗時(shí)：{sync_test(10):.2f}秒")
print("\n異步版，10個(gè)請(qǐng)求：")
print(f"耗時(shí)：{asyncio.run(async_test(10)):.2f}秒")

在我的測(cè)試中，同步版大約需要10秒（順序執(zhí)行，每個(gè)1秒），而異步版大約只要1秒多（并發(fā)執(zhí)行）。當(dāng)請(qǐng)求數(shù)量增加到100時(shí)，差異會(huì)更加明顯。

現(xiàn)在開始用協(xié)程還太早嗎？

如果你的項(xiàng)目主要是CPU密集型計(jì)算（比如數(shù)據(jù)分析、圖像處理），協(xié)程帶來(lái)的提升可能有限。但如果是Web服務(wù)、爬蟲、聊天機(jī)器人這類I/O密集的應(yīng)用，協(xié)程幾乎成了標(biāo)配。

學(xué)習(xí)曲線呢？確實(shí)需要一點(diǎn)時(shí)間適應(yīng)。從同步思維切換到異步思維，就像從單線程轉(zhuǎn)到多線程一樣，需要重新考慮程序的組織方式。不過(guò)一旦掌握，代碼的性能和可讀性都會(huì)有很大改善。

Python的異步生態(tài)已經(jīng)相當(dāng)成熟了。Web框架有FastAPI（性能強(qiáng)悍，還自動(dòng)生成API文檔）、Sanic；數(shù)據(jù)庫(kù)有aiomysql、asyncpg；HTTP客戶端有aiohttp；甚至機(jī)器學(xué)習(xí)領(lǐng)域也開始出現(xiàn)異步支持。

下一步該學(xué)什么？

今天我們從協(xié)程的基本概念講到了實(shí)際應(yīng)用，但這只是異步編程的起點(diǎn)。下次我們會(huì)深入更多實(shí)用話題：

1. 協(xié)程間的通信：多個(gè)協(xié)程怎么安全地共享數(shù)據(jù)？asyncio.Queue怎么用？
2. 同步原語(yǔ)：協(xié)程版本的鎖（Lock）、信號(hào)量（Semaphore）、事件（Event）是什么？
3. 錯(cuò)誤處理：協(xié)程里的異常怎么捕獲和處理？
4. 與線程/進(jìn)程結(jié)合：如何在協(xié)程里調(diào)用同步代碼？怎么利用多核CPU？
5. 實(shí)際項(xiàng)目結(jié)構(gòu)：大型異步項(xiàng)目該怎么組織代碼？

這些話題我會(huì)在下一篇文章中詳細(xì)講解。如果你已經(jīng)躍躍欲試，建議先從一個(gè)小項(xiàng)目開始，比如寫個(gè)異步爬蟲，或者用FastAPI搭個(gè)簡(jiǎn)單的Web服務(wù)。

你平時(shí)寫代碼時(shí)，遇到過(guò)哪些適合用協(xié)程解決的場(chǎng)景？或者對(duì)協(xié)程的哪些部分感到困惑？歡迎在評(píng)論區(qū)聊聊你的經(jīng)驗(yàn)或問(wèn)題，我們一起探討。

到此這篇關(guān)于Python異步編程入門：協(xié)程到底是什么？的文章就介紹到這了,更多相關(guān)Python協(xié)程內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！