Python異步編程asyncio的核心用法與避坑指南

更新時間：2026年03月26日 08:22:10 作者：傻啦嘿喲

本文介紹了Python異步編程的基本概念、核心價值及其實(shí)現(xiàn)機(jī)制,并詳細(xì)解釋了如何通過async和await關(guān)鍵字來編寫異步代碼,同時列舉了初學(xué)者常犯的錯誤及避免這些問題的方法,需要的朋友可以參考下

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

這種感覺不怪你。異步編程確實(shí)和Python其他部分不太一樣，但一旦跨過那道坎，你會發(fā)現(xià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"總耗時:?{time.time()?-?start}")

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

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

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

async和await到底在干什么

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

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

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

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

來看一個完整的例子：

import?asyncio

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

async?def?main():
????#?創(chuàng)建三個任務(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é)束
總耗時約3秒

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

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

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

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

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

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

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

新手最容易踩的坑

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

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

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

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

坑二：忘記加await

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

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

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

def?sync_function():
????#?錯誤：不能在同步函數(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)致資源耗盡

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

雖然協(xié)程比線程輕量很多，但也不是無限創(chuàng)建的。幾萬個協(xié)程同時存在，內(nèi)存和調(diào)度開銷還是會很大。可以用信號量（Semaphore）來控制并發(fā)數(shù)量。

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

在實(shí)際項目中，你通常不會把所有任務(wù)一次性丟進(jìn)事件循環(huán)。以爬蟲為例，同時發(fā)幾百個請求，目標(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:??#?獲取信號量，控制并發(fā)數(shù)
????????async?with?session.get(url)?as?response:
????????????return?await?response.text()

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

asyncio.run(main())

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

超時處理是保命技能

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

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

import?asyncio

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

async?def?worker(name,?queue,?session):
????"""工作協(xié)程：從隊列取URL，抓取頁面"""
????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}?成功，長度?{len(content)}")
????????????????#?這里可以存儲數(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放入隊列
????for?url?in?urls:
????????await?queue.put(url)
????
????async?with?aiohttp.ClientSession()?as?session:
????????#?啟動5個工作協(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())

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