你的爬蟲程序又卡住了。

不是代碼寫錯(cuò)了，是請(qǐng)求發(fā)得太慢。一萬條數(shù)據(jù)要爬，一條一秒，兩個(gè)多小時(shí)才能跑完。你試著用多線程，結(jié)果IP被封得更快了。線程開太多，系統(tǒng)資源也扛不住。你聽說過異步，聽說它能用單線程搞定高并發(fā)，但看著async和await這兩個(gè)關(guān)鍵字，總覺得像在寫另一門語(yǔ)言。

這種感覺不怪你。異步編程確實(shí)和Python其他部分不太一樣，但一旦跨過那道坎，你會(huì)發(fā)現(xiàn)它沒有想象中那么難。

異步到底在解決什么問題

先看一個(gè)很簡(jiǎn)單的程序：

import?time

def?task(name,?seconds):
????print(f"{name}?開始")
????time.sleep(seconds)
????print(f"{name}?結(jié)束")

start?=?time.time()
task("任務(wù)1",?2)
task("任務(wù)2",?2)
print(f"總耗時(shí):?{time.time()?-?start}")

運(yùn)行結(jié)果是：任務(wù)1開始、等待兩秒、任務(wù)1結(jié)束、任務(wù)2開始、再等兩秒、任務(wù)2結(jié)束，總耗時(shí)4秒。

問題出在time.sleep這里。當(dāng)程序執(zhí)行到sleep時(shí)，CPU其實(shí)什么事都沒干，就傻等著時(shí)間過去。這段時(shí)間本可以用來處理其他任務(wù)，但同步代碼不允許這樣做——它是一條路走到黑的，沒執(zhí)行完當(dāng)前函數(shù)，絕不會(huì)跳到下一行。

異步要解決的就是這個(gè)“等待浪費(fèi)”的問題。當(dāng)某個(gè)操作需要等待時(shí)，讓程序暫時(shí)離開，去做別的事，等那個(gè)操作準(zhǔn)備好了再回來繼續(xù)。

async和await到底在干什么

Python的異步編程基于一個(gè)叫做“事件循環(huán)”的東西。你可以把它理解成一個(gè)調(diào)度中心，管理著所有待執(zhí)行的任務(wù)。當(dāng)一個(gè)任務(wù)遇到需要等待的操作時(shí)，它會(huì)告訴事件循環(huán)：“我先去忙別的，好了叫我”，然后事件循環(huán)就會(huì)切換到下一個(gè)可以執(zhí)行的任務(wù)。

async def用來定義一個(gè)異步函數(shù)，也叫協(xié)程。它和普通函數(shù)的區(qū)別在于，調(diào)用它并不會(huì)立即執(zhí)行，而是返回一個(gè)協(xié)程對(duì)象。

async?def?hello():
????print("Hello")
????await?asyncio.sleep(1)
????print("World")

await用來等待一個(gè)異步操作完成。當(dāng)程序執(zhí)行到await時(shí)，它會(huì)暫時(shí)離開這個(gè)函數(shù)，讓事件循環(huán)去處理其他任務(wù)。等asyncio.sleep(1)這個(gè)操作完成了，事件循環(huán)才會(huì)回來繼續(xù)執(zhí)行后面的print("World")。

來看一個(gè)完整的例子：

import?asyncio

async?def?task(name,?seconds):
????print(f"{name}?開始")
????await?asyncio.sleep(seconds)
????print(f"{name}?結(jié)束")

async?def?main():
????#?創(chuàng)建三個(gè)任務(wù)，但還沒有執(zhí)行
????tasks?=?[
????????asyncio.create_task(task("任務(wù)1",?2)),
????????asyncio.create_task(task("任務(wù)2",?1)),
????????asyncio.create_task(task("任務(wù)3",?3))
????]
????#?等待所有任務(wù)完成
????await?asyncio.gather(*tasks)

asyncio.run(main())

運(yùn)行結(jié)果：

任務(wù)1?開始
任務(wù)2?開始
任務(wù)3?開始
任務(wù)2?結(jié)束
任務(wù)1?結(jié)束
任務(wù)3?結(jié)束
總耗時(shí)約3秒

三個(gè)任務(wù)幾乎是同時(shí)開始的，總共只用了最慢那個(gè)任務(wù)的時(shí)間。這就是異步的核心價(jià)值：在等待的時(shí)間里做別的事。

事件循環(huán)是怎么運(yùn)轉(zhuǎn)的

理解事件循環(huán)的工作機(jī)制，能幫你避開大部分異步編程的坑。

事件循環(huán)本質(zhì)上是一個(gè)無限循環(huán)，它維護(hù)著兩個(gè)隊(duì)列：就緒隊(duì)列和等待隊(duì)列。就緒隊(duì)列里放著可以立即執(zhí)行的任務(wù)，等待隊(duì)列里放著正在等待某個(gè)事件（比如網(wǎng)絡(luò)響應(yīng)、定時(shí)器到期）的任務(wù)。

每一次循環(huán)，事件循環(huán)會(huì)從就緒隊(duì)列里取出一個(gè)任務(wù)執(zhí)行。當(dāng)任務(wù)執(zhí)行到await時(shí)，它會(huì)把自己掛起，并告訴事件循環(huán)它正在等待什么。事件循環(huán)就把這個(gè)任務(wù)放到等待隊(duì)列里，然后繼續(xù)處理就緒隊(duì)列中的下一個(gè)任務(wù)。

當(dāng)?shù)却?duì)列里的某個(gè)任務(wù)等的事件發(fā)生了（比如定時(shí)器到期），事件循環(huán)就會(huì)把它移回就緒隊(duì)列，等待下一次被調(diào)度執(zhí)行。

這就是為什么異步程序能在一個(gè)線程里實(shí)現(xiàn)并發(fā)：它不是同時(shí)做多件事，而是在一件事等待的時(shí)候，去做另一件事。

新手最容易踩的坑

坑一：在異步函數(shù)里用了同步阻塞操作

async?def?bad_example():
????#?錯(cuò)誤：用了?time.sleep?而不是?asyncio.sleep
????time.sleep(5)??#?這會(huì)阻塞整個(gè)事件循環(huán)
????return?"done"

time.sleep是同步阻塞的，它會(huì)卡住當(dāng)前線程，事件循環(huán)在這5秒內(nèi)完全無法工作，所有任務(wù)都會(huì)被堵住。正確的做法是用await asyncio.sleep(5)。

同樣的道理，如果你在異步代碼里使用了requests庫(kù)發(fā)HTTP請(qǐng)求，它也會(huì)阻塞事件循環(huán)。應(yīng)該用aiohttp這類異步HTTP客戶端。

坑二：忘記加await

async?def?wrong():
????coro?=?some_async_function()??#?這返回的是協(xié)程對(duì)象，不是執(zhí)行結(jié)果
????print(coro)??#?打印?<coroutine?object...>
????#?協(xié)程對(duì)象沒有被?await，它永遠(yuǎn)不會(huì)執(zhí)行

協(xié)程對(duì)象被創(chuàng)建后，必須被await或者被asyncio.create_task()調(diào)度，否則它不會(huì)執(zhí)行。這是異步編程新手最容易忽略的地方。

坑三：在同步代碼里調(diào)用異步函數(shù)

def?sync_function():
????#?錯(cuò)誤：不能在同步函數(shù)里直接?await
????result?=?await?async_function()??#?SyntaxError

await只能在async def定義的函數(shù)內(nèi)部使用。如果你想在同步代碼里執(zhí)行異步函數(shù)，需要用asyncio.run()或者asyncio.create_task()配合事件循環(huán)。

坑四：創(chuàng)建了太多任務(wù)導(dǎo)致資源耗盡

#?錯(cuò)誤：一下子創(chuàng)建幾萬個(gè)任務(wù)
for?i?in?range(100000):
????asyncio.create_task(heavy_task(i))
await?asyncio.gather(*tasks)??#?可能內(nèi)存爆炸

雖然協(xié)程比線程輕量很多，但也不是無限創(chuàng)建的。幾萬個(gè)協(xié)程同時(shí)存在，內(nèi)存和調(diào)度開銷還是會(huì)很大?？梢杂眯盘?hào)量（Semaphore）來控制并發(fā)數(shù)量。

控制并發(fā)數(shù)的正確姿勢(shì)

在實(shí)際項(xiàng)目中，你通常不會(huì)把所有任務(wù)一次性丟進(jìn)事件循環(huán)。以爬蟲為例，同時(shí)發(fā)幾百個(gè)請(qǐng)求，目標(biāo)網(wǎng)站可能直接把你IP封了，你自己的網(wǎng)絡(luò)也可能扛不住。

用asyncio.Semaphore可以輕松控制并發(fā)數(shù)量：

import?asyncio
import?aiohttp

async?def?fetch(session,?url,?semaphore):
????async?with?semaphore:??#?獲取信號(hào)量，控制并發(fā)數(shù)
????????async?with?session.get(url)?as?response:
????????????return?await?response.text()

async?def?main():
????urls?=?["http://example.com"]?*?100??#?100個(gè)URL
????semaphore?=?asyncio.Semaphore(10)??#?限制最多同時(shí)10個(gè)請(qǐng)求
????
????async?with?aiohttp.ClientSession()?as?session:
????????tasks?=?[fetch(session,?url,?semaphore)?for?url?in?urls]
????????results?=?await?asyncio.gather(*tasks)
????
????print(f"抓取完成，共{len(results)}個(gè)頁(yè)面")

asyncio.run(main())

信號(hào)量的工作原理很簡(jiǎn)單：它內(nèi)部維護(hù)一個(gè)計(jì)數(shù)器。每次async with semaphore會(huì)嘗試減少計(jì)數(shù)器，如果計(jì)數(shù)器大于0，就允許進(jìn)入；如果等于0，就等待直到有任務(wù)釋放信號(hào)量。這樣就能精確控制并發(fā)數(shù)，既充分利用資源，又不至于把對(duì)方服務(wù)器打趴下。

超時(shí)處理是保命技能

異步程序里，某個(gè)任務(wù)卡住會(huì)影響整個(gè)事件循環(huán)嗎？不會(huì)，因?yàn)?code>await是讓出控制權(quán)的。但如果一個(gè)任務(wù)內(nèi)部有死循環(huán)或者一直沒遇到await，它確實(shí)會(huì)霸占事件循環(huán)，導(dǎo)致其他任務(wù)無法執(zhí)行。

更常見的問題是網(wǎng)絡(luò)請(qǐng)求一直不返回。這時(shí)候就需要超時(shí)控制：

import?asyncio

async?def?fetch_with_timeout():
????try:
????????#?設(shè)置5秒超時(shí)
????????result?=?await?asyncio.wait_for(slow_operation(),?timeout=5)
????????return?result
????except?asyncio.TimeoutError:
????????print("操作超時(shí)")
????????return?None

async?def?slow_operation():
????await?asyncio.sleep(10)??#?模擬慢操作
????return?"數(shù)據(jù)"

asyncio.wait_for會(huì)給一個(gè)協(xié)程加上時(shí)間限制。超時(shí)后它會(huì)拋出asyncio.TimeoutError，你可以捕獲它做降級(jí)處理。

還有一個(gè)更靈活的工具是asyncio.gather的return_exceptions參數(shù)：

results?=?await?asyncio.gather(
????task1(),
????task2(),
????task3(),
????return_exceptions=True??#?異常不會(huì)中斷，而是作為結(jié)果返回
)

for?result?in?results:
????if?isinstance(result,?Exception):
????????print(f"任務(wù)失敗:?{result}")
????else:
????????print(f"任務(wù)成功:?{result}")

這樣單個(gè)任務(wù)失敗不會(huì)導(dǎo)致整個(gè)gather崩潰，你可以優(yōu)雅地處理每個(gè)任務(wù)的結(jié)果。

異步和并發(fā)的本質(zhì)區(qū)別

很多人會(huì)把異步和多線程混淆，覺得它們都能“同時(shí)”做多件事。但理解它們的區(qū)別，能幫你做出更合理的技術(shù)選型。

多線程是操作系統(tǒng)層面的并發(fā)。每個(gè)線程都有自己的?？臻g，線程切換由操作系統(tǒng)調(diào)度，開銷較大。Python因?yàn)橛蠫IL（全局解釋器鎖），多線程在CPU密集型任務(wù)上反而更慢，但在IO密集型任務(wù)上依然有用。

異步是單線程內(nèi)的并發(fā)。所有協(xié)程共享同一個(gè)線程，切換發(fā)生在await的時(shí)候，開銷極小。一個(gè)事件循環(huán)可以輕松處理上萬協(xié)程，而開上萬個(gè)線程基本不可能。

選哪個(gè)？IO密集型任務(wù)（網(wǎng)絡(luò)請(qǐng)求、文件讀寫、數(shù)據(jù)庫(kù)查詢）用異步，代碼更簡(jiǎn)潔，資源消耗更少。CPU密集型任務(wù)（計(jì)算、加密、壓縮）用多進(jìn)程，或者把計(jì)算部分交給專門的進(jìn)程池，異步只負(fù)責(zé)調(diào)度。

實(shí)際項(xiàng)目中的異步架構(gòu)

真實(shí)項(xiàng)目很少只用一個(gè)asyncio.run()就完事。更常見的做法是分層設(shè)計(jì)：

底層是異步IO操作，用aiohttp、aiomysql、aioredis這些異步庫(kù)。中間層是業(yè)務(wù)邏輯，用async def定義協(xié)程，處理數(shù)據(jù)轉(zhuǎn)換、錯(cuò)誤重試、超時(shí)控制。最上層是調(diào)度層，管理任務(wù)隊(duì)列、控制并發(fā)、監(jiān)控執(zhí)行狀態(tài)。

一個(gè)典型的數(shù)據(jù)采集任務(wù)大概是這樣的：

import?asyncio
import?aiohttp
from?asyncio?import?Queue

async?def?worker(name,?queue,?session):
????"""工作協(xié)程：從隊(duì)列取URL，抓取頁(yè)面"""
????while?True:
????????url?=?await?queue.get()
????????try:
????????????async?with?session.get(url,?timeout=aiohttp.ClientTimeout(total=10))?as?resp:
????????????????content?=?await?resp.text()
????????????????print(f"{name}?抓取?{url}?成功，長(zhǎng)度?{len(content)}")
????????????????#?這里可以存儲(chǔ)數(shù)據(jù)
????????except?Exception?as?e:
????????????print(f"{name}?抓取?{url}?失敗:?{e}")
????????finally:
????????????queue.task_done()

async?def?main():
????urls?=?[f"http://example.com/page/{i}"?for?i?in?range(1000)]
????queue?=?Queue()
????
????#?把所有URL放入隊(duì)列
????for?url?in?urls:
????????await?queue.put(url)
????
????async?with?aiohttp.ClientSession()?as?session:
????????#?啟動(dòng)5個(gè)工作協(xié)程
????????workers?=?[asyncio.create_task(worker(f"worker-{i}",?queue,?session))?
???????????????????for?i?in?range(5)]
????????
????????#?等待所有任務(wù)完成
????????await?queue.join()
????????
????????#?取消工作協(xié)程
????????for?w?in?workers:
????????????w.cancel()

asyncio.run(main())

隊(duì)列模式的好處是解耦了任務(wù)生產(chǎn)者和消費(fèi)者，可以靈活調(diào)整并發(fā)數(shù)量，也方便做斷點(diǎn)續(xù)傳、失敗重試這些復(fù)雜邏輯。

寫在最后

異步編程剛接觸時(shí)確實(shí)有點(diǎn)繞，但核心思想很簡(jiǎn)單：別閑著，等的時(shí)候去干點(diǎn)別的。async定義協(xié)程，await讓出控制權(quán)，事件循環(huán)負(fù)責(zé)調(diào)度。記住這三件事，你就已經(jīng)入門了。

那些坑，說到底都是忘了“異步代碼里不能有阻塞操作”這條原則。遇到問題的時(shí)候，先檢查是不是用了同步庫(kù)，是不是忘記await了，是不是并發(fā)數(shù)設(shè)置得太高。

異步Python這幾年越來越成熟。aiohttp、asyncpg、FastAPI這些生態(tài)組件已經(jīng)足夠支撐大型項(xiàng)目。如果你還在用多線程處理IO密集型任務(wù)，不妨試試異步。同樣的機(jī)器資源，異步往往能扛住更高的并發(fā)，代碼邏輯也更清晰。

寫異步代碼像是在指揮一支交響樂團(tuán)——每個(gè)樂器都在自己的節(jié)奏上演奏，你不需要盯著每個(gè)人，只需要把握好整體進(jìn)度。一旦適應(yīng)了這種編程思維，你會(huì)發(fā)現(xiàn)，原來高并發(fā)也可以寫得這么優(yōu)雅。

以上就是Python異步編程asyncio的核心用法與避坑指南的詳細(xì)內(nèi)容，更多關(guān)于Python異步編程asyncio的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

最新評(píng)論