秘密：源碼拆解實(shí)戰(zhàn)項(xiàng)目中的高并發(fā)陷阱)
Python asyncio 毫無(wú)秘密：源碼拆解實(shí)戰(zhàn)項(xiàng)目中的高并發(fā)陷阱
配置環(huán)境就卡半天，跑個(gè)實(shí)戰(zhàn)項(xiàng)目直接內(nèi)存泄漏？別急，這往往不是代碼寫(xiě)錯(cuò)，而是你根本沒(méi)搞懂 asyncio 的底層邏輯。很多轉(zhuǎn)崗做后端的同學(xué)，面試時(shí)能把事件循環(huán)講得頭頭是道，一到真實(shí)業(yè)務(wù)場(chǎng)景，并發(fā)量稍微一上來(lái)，CPU 飆升，連接池耗盡，整個(gè)人就懵了。
我在 Stack Overflow 上見(jiàn)過(guò)太多類(lèi)似問(wèn)題，90% 的答案都在說(shuō)“檢查阻塞調(diào)用”。但這句話太泛了，到底哪里阻塞了？為什么 await 了還是卡？今天咱們不背八股文，直接扒開(kāi) CPython 3.10+ 的 asyncio 核心源碼，看看事件循環(huán)到底是怎么調(diào)度的。你會(huì)明白，那些看似玄學(xué)的性能問(wèn)題，其實(shí)都藏在幾十行代碼里。
入口定位：從 run_until_complete 開(kāi)始
很多教程讓你直接 asyncio.run(main())，然后就沒(méi)了。這就像教你開(kāi)車(chē)只教了踩油門(mén)，沒(méi)教你看路況。我們得從 asyncio/runners.py 入手，看看 run() 到底干了什么。
# 源碼片段 1：asyncio/runners.py (簡(jiǎn)化版)
def run(main, *, debug=False, loop_factory=None):if coroutines.iscoroutine(main):coro = mainelse:raise TypeError('an asyncio coroutine is required')if events._get_running_loop() is not None:raise RuntimeError(asyncio.run() cannot be called from a running event loop)with Runner(debug=debug, loop_factory=loop_factory) as runner:return runner.run(coro)逐行拆解：iscoroutine(main)：檢查傳入的是不是協(xié)程對(duì)象。很多人傳函數(shù)而不是協(xié)程，這里直接拋異常，這是最常見(jiàn)的低級(jí)錯(cuò)誤。
_get_running_loop()：這是關(guān)鍵。如果當(dāng)前線程已經(jīng)有運(yùn)行中的事件循環(huán)，直接報(bào)錯(cuò)。這解釋了為什么你在 Jupyter Notebook 或某些框架里嵌套調(diào)用 asyncio.run() 會(huì)炸。
Runner 上下文管理器：它負(fù)責(zé)創(chuàng)建新的事件循環(huán)，運(yùn)行主協(xié)程，并在結(jié)束后清理資源。重點(diǎn)在于“清理”，很多內(nèi)存泄漏就出在這里，事件循環(huán)沒(méi)關(guān)干凈，定時(shí)器或任務(wù)殘留。核心片段：事件循環(huán)的調(diào)度心臟
真正的魔法在 base_events.py 的 _run_once 方法。這是 asyncio 的心跳，每次 await 讓出控制權(quán)，或者 I/O 就緒，都會(huì)走到這里。
# 源碼片段 2：asyncio/base_events.py (簡(jiǎn)化版)
def _run_once(self):if self._stopping:raise RuntimeError('Event loop stopped')timer_handle = Noneif self._scheduled:now = self.time()while self._scheduled:handle = self._scheduled[0]if handle._when = now:breakhandle = heapq.heappop(self._scheduled)self._ready.append(handle)# 處理 I/O 就緒事件event_list = self._selector.select(timeout)for key, events in event_list:callback = key.dataself._add_callback(callback)# 執(zhí)行 ready 隊(duì)列中的任務(wù)ntodo = len(self._ready)for i in range(ntodo):handle = self._ready.popleft()if handle._cancelled:continuehandle._run()逐行拆解：self._scheduled：這是一個(gè)最小堆，存放 call_later 注冊(cè)的任務(wù)。heapq.heappop 保證我們總是先處理最早到期的任務(wù)，時(shí)間復(fù)雜度是 O(log n)。
self._selector.select(timeout)：這是底層的 select/epoll/kqueue 封裝。timeout 的計(jì)算非常講究，它取的是“下一個(gè)定時(shí)器觸發(fā)時(shí)間”和“最大 I/O 等待時(shí)間”的較小值。如果算錯(cuò)了，要么 CPU 空轉(zhuǎn)，要么響應(yīng)延遲。
handle._run()：這里執(zhí)行的是回調(diào)函數(shù)。注意，asyncio 是單線程的，所以這里的 _run 必須是純 CPU 計(jì)算或立即返回 I/O 就緒狀態(tài)。如果這里出現(xiàn)阻塞調(diào)用（比如 time.sleep(1)），整個(gè)事件循環(huán)就卡死了。這就是為什么在 async 函數(shù)里不能用 requests，必須用 aiohttp。設(shè)計(jì)思想：為什么是單線程高并發(fā)？
理解了代碼，再回看設(shè)計(jì)思想。asyncio 的核心假設(shè)是：大部分時(shí)間線程都在等待 I/O，CPU 是空閑的。
傳統(tǒng)多線程模型中，每個(gè)請(qǐng)求一個(gè)線程，上下文切換開(kāi)銷(xiāo)大，內(nèi)存占用高。asyncio 用協(xié)程（用戶態(tài)線程）替代，切換成本極低（微秒級(jí)）。但代價(jià)是：一旦有同步阻塞代碼，整個(gè)進(jìn)程就停擺了。
這就是為什么在實(shí)戰(zhàn)項(xiàng)目中，我們常說(shuō)“異步不阻塞”。這不是口號(hào)，是生存法則。很多轉(zhuǎn)崗前端轉(zhuǎn)后端的同學(xué)，習(xí)慣用 setTimeout 來(lái)模擬異步，這在 Python 里是災(zāi)難。asyncio 的協(xié)作式多任務(wù)，要求每個(gè)協(xié)程在合適的時(shí)候主動(dòng)讓出控制權(quán)（await）。如果你不讓?zhuān)瑒e人就得等著，一鍋端。
手寫(xiě)簡(jiǎn)化版：一個(gè)迷你事件循環(huán)
為了加深理解，我們手寫(xiě)一個(gè)極簡(jiǎn)版的 EventLoop。代碼不長(zhǎng)，但涵蓋了核心邏輯。
import heapq
import timeclass MiniEventLoop:def __init__(self):self._ready = [] # 就緒隊(duì)列self._scheduled = [] # 定時(shí)器堆self._counter = 0 # 用于排序的穩(wěn)定標(biāo)識(shí)def call_later(self, delay, callback, *args):when = time.time() + delayheapq.heappush(self._scheduled, (when, self._counter, callback, args))self._counter += 1def run_forever(self):while True:# 1. 處理定時(shí)器now = time.time()while self._scheduled and self._scheduled[0][0] = now:when, _, callback, args = heapq.heappop(self._scheduled)self._ready.append(callback)# 2. 如果沒(méi)有就緒任務(wù)且沒(méi)有定時(shí)器，退出（實(shí)際中應(yīng)阻塞等待 I/O）if not self._ready and not self._scheduled:break# 3. 執(zhí)行就緒任務(wù)if self._ready:callback = self._ready.pop(0)callback(*callback_args)# 使用示例
loop = MiniEventLoop()
loop.call_later(1, lambda: print(Hello after 1s))
loop.call_later(0.5, lambda: print(Hello after 0.5s))
loop.run_forever()這個(gè)簡(jiǎn)化版少了 I/O 多路復(fù)用，但保留了定時(shí)器和就緒隊(duì)列的核心邏輯。你可以看到，asyncio 本質(zhì)上就是一個(gè)帶定時(shí)器的任務(wù)隊(duì)列調(diào)度器。在實(shí)際項(xiàng)目中，aiohttp 的網(wǎng)絡(luò)請(qǐng)求回調(diào)，最終也是通過(guò) loop.call_soon 進(jìn)入 _ready 隊(duì)列，等待被執(zhí)行。
應(yīng)用場(chǎng)景：實(shí)戰(zhàn)項(xiàng)目中的避坑指南
回到實(shí)戰(zhàn)項(xiàng)目。假設(shè)你在做一個(gè)高并發(fā)的 WebSocket 聊天服務(wù)，用戶數(shù)上萬(wàn)。常見(jiàn)坑點(diǎn)有這三個(gè)：數(shù)據(jù)庫(kù)同步驅(qū)動(dòng)：asyncpg 是異步的，psycopg2 是同步的。如果你在 async 函數(shù)里用 psycopg2 查詢(xún)，哪怕只查 10ms，也會(huì)阻塞事件循環(huán) 10ms。如果有 1000 個(gè)并發(fā)，延遲就是 10 秒。解決方案：用 asyncio.to_thread 把同步調(diào)用扔到線程池，或者換用異步驅(qū)動(dòng)。任務(wù)未取消：用戶斷開(kāi)連接時(shí)，如果沒(méi)取消對(duì)應(yīng)的協(xié)程任務(wù)，它會(huì)繼續(xù)運(yùn)行，占用內(nèi)存。務(wù)必使用 try/except CancelledError 并清理資源。信號(hào)量濫用：用 asyncio.Semaphore 控制并發(fā)數(shù)是對(duì)的，但注意它的釋放必須在 finally 塊中。如果協(xié)程異常退出，信號(hào)量沒(méi)釋放，后續(xù)請(qǐng)求全部阻塞。這些坑，Stack Overflow 上每天都有人問(wèn)。但如果你懂 _run_once 的執(zhí)行流程，你就知道問(wèn)題出在哪。不要迷信框架，要理解底層。
你在項(xiàng)目里踩過(guò)這個(gè)坑嗎？評(píng)論區(qū)聊聊 Python asyncio 毫無(wú)秘密：源碼拆解實(shí)戰(zhàn)項(xiàng)目中的高并發(fā)陷阱
Python asyncio 毫無(wú)秘密：源碼拆解實(shí)戰(zhàn)項(xiàng)目中的高并發(fā)陷阱
配置環(huán)境就卡半天，跑個(gè)實(shí)戰(zhàn)項(xiàng)目直接內(nèi)存泄漏？別急，這往往不是代碼寫(xiě)錯(cuò)，而是你根本沒(méi)搞懂 asyncio 的底層邏輯。很多轉(zhuǎn)崗做后端的同學(xué)，面試時(shí)能把事件循環(huán)講得頭頭是道，一到真實(shí)業(yè)務(wù)場(chǎng)景，并發(fā)量稍微一上來(lái)，CPU 飆升，連接池耗盡，整個(gè)人就懵了。
我在 Stack Overflow 上見(jiàn)過(guò)太多類(lèi)似問(wèn)題，90% 的答案都在說(shuō)“檢查阻塞調(diào)用”。但這句話太泛了，到底哪里阻塞了？為什么 await 了還是卡？今天咱們不背八股文，直接扒開(kāi) CPython 3.10+ 的 asyncio 核心源碼，看看事件循環(huán)到底是怎么調(diào)度的。你會(huì)明白，那些看似玄學(xué)的性能問(wèn)題，其實(shí)都藏在幾十行代碼里。
入口定位：從 run_until_complete 開(kāi)始
很多教程讓你直接 asyncio.run(main())，然后就沒(méi)了。這就像教你開(kāi)車(chē)只教了踩油門(mén)，沒(méi)教你看路況。我們得從 asyncio/runners.py 入手，看看 run() 到底干了什么。
# 源碼片段 1：asyncio/runners.py (簡(jiǎn)化版)
def run(main, *, debug=False, loop_factory=None):if coroutines.iscoroutine(main):coro = mainelse:raise TypeError('an asyncio coroutine is required')if events._get_running_loop() is not None:raise RuntimeError(asyncio.run() cannot be called from a running event loop)with Runner(debug=debug, loop_factory=loop_factory) as runner:return runner.run(coro)逐行拆解：iscoroutine(main)：檢查傳入的是不是協(xié)程對(duì)象。很多人傳函數(shù)而不是協(xié)程，這里直接拋異常，這是最常見(jiàn)的低級(jí)錯(cuò)誤。
_get_running_loop()：這是關(guān)鍵。如果當(dāng)前線程已經(jīng)有運(yùn)行中的事件循環(huán)，直接報(bào)錯(cuò)。這解釋了為什么你在 Jupyter Notebook 或某些框架里嵌套調(diào)用 asyncio.run() 會(huì)炸。
Runner 上下文管理器：它負(fù)責(zé)創(chuàng)建新的事件循環(huán)，運(yùn)行主協(xié)程，并在結(jié)束后清理資源。重點(diǎn)在于“清理”，很多內(nèi)存泄漏就出在這里，事件循環(huán)沒(méi)關(guān)干凈，定時(shí)器或任務(wù)殘留。核心片段：事件循環(huán)的調(diào)度心臟
真正的魔法在 base_events.py 的 _run_once 方法。這是 asyncio 的心跳，每次 await 讓出控制權(quán)，或者 I/O 就緒，都會(huì)走到這里。
# 源碼片段 2：asyncio/base_events.py (簡(jiǎn)化版)
def _run_once(self):if self._stopping:raise RuntimeError('Event loop stopped')timer_handle = Noneif self._scheduled:now = self.time()while self._scheduled:handle = self._scheduled[0]if handle._when = now:breakhandle = heapq.heappop(self._scheduled)self._ready.append(handle)# 處理 I/O 就緒事件event_list = self._selector.select(timeout)for key, events in event_list:callback = key.dataself._add_callback(callback)# 執(zhí)行 ready 隊(duì)列中的任務(wù)ntodo = len(self._ready)for i in range(ntodo):handle = self._ready.popleft()if handle._cancelled:continuehandle._run()逐行拆解：self._scheduled：這是一個(gè)最小堆，存放 call_later 注冊(cè)的任務(wù)。heapq.heappop 保證我們總是先處理最早到期的任務(wù)，時(shí)間復(fù)雜度是 O(log n)。
self._selector.select(timeout)：這是底層的 select/epoll/kqueue 封裝。timeout 的計(jì)算非常講究，它取的是“下一個(gè)定時(shí)器觸發(fā)時(shí)間”和“最大 I/O 等待時(shí)間”的較小值。如果算錯(cuò)了，要么 CPU 空轉(zhuǎn)，要么響應(yīng)延遲。
handle._run()：這里執(zhí)行的是回調(diào)函數(shù)。注意，asyncio 是單線程的，所以這里的 _run 必須是純 CPU 計(jì)算或立即返回 I/O 就緒狀態(tài)。如果這里出現(xiàn)阻塞調(diào)用（比如 time.sleep(1)），整個(gè)事件循環(huán)就卡死了。這就是為什么在 async 函數(shù)里不能用 requests，必須用 aiohttp。設(shè)計(jì)思想：為什么是單線程高并發(fā)？
理解了代碼，再回看設(shè)計(jì)思想。asyncio 的核心假設(shè)是：大部分時(shí)間線程都在等待 I/O，CPU 是空閑的。
傳統(tǒng)多線程模型中，每個(gè)請(qǐng)求一個(gè)線程，上下文切換開(kāi)銷(xiāo)大，內(nèi)存占用高。asyncio 用協(xié)程（用戶態(tài)線程）替代，切換成本極低（微秒級(jí)）。但代價(jià)是：一旦有同步阻塞代碼，整個(gè)進(jìn)程就停擺了。
這就是為什么在實(shí)戰(zhàn)項(xiàng)目中，我們常說(shuō)“異步不阻塞”。這不是口號(hào)，是生存法則。很多轉(zhuǎn)崗前端轉(zhuǎn)后端的同學(xué)，習(xí)慣用 setTimeout 來(lái)模擬異步，這在 Python 里是災(zāi)難。asyncio 的協(xié)作式多任務(wù)，要求每個(gè)協(xié)程在合適的時(shí)候主動(dòng)讓出控制權(quán)（await）。如果你不讓?zhuān)瑒e人就得等著，一鍋端。
手寫(xiě)簡(jiǎn)化版：一個(gè)迷你事件循環(huán)
為了加深理解，我們手寫(xiě)一個(gè)極簡(jiǎn)版的 EventLoop。代碼不長(zhǎng)，但涵蓋了核心邏輯。
import heapq
import timeclass MiniEventLoop:def __init__(self):self._ready = [] # 就緒隊(duì)列self._scheduled = [] # 定時(shí)器堆self._counter = 0 # 用于排序的穩(wěn)定標(biāo)識(shí)def call_later(self, delay, callback, *args):when = time.time() + delayheapq.heappush(self._scheduled, (when, self._counter, callback, args))self._counter += 1def run_forever(self):while True:# 1. 處理定時(shí)器now = time.time()while self._scheduled and self._scheduled[0][0] = now:when, _, callback, args = heapq.heappop(self._scheduled)self._ready.append(callback)# 2. 如果沒(méi)有就緒任務(wù)且沒(méi)有定時(shí)器，退出（實(shí)際中應(yīng)阻塞等待 I/O）if not self._ready and not self._scheduled:break# 3. 執(zhí)行就緒任務(wù)if self._ready:callback = self._ready.pop(0)callback(*callback_args)# 使用示例
loop = MiniEventLoop()
loop.call_later(1, lambda: print(Hello after 1s))
loop.call_later(0.5, lambda: print(Hello after 0.5s))
loop.run_forever()這個(gè)簡(jiǎn)化版少了 I/O 多路復(fù)用，但保留了定時(shí)器和就緒隊(duì)列的核心邏輯。你可以看到，asyncio 本質(zhì)上就是一個(gè)帶定時(shí)器的任務(wù)隊(duì)列調(diào)度器。在實(shí)際項(xiàng)目中，aiohttp 的網(wǎng)絡(luò)請(qǐng)求回調(diào)，最終也是通過(guò) loop.call_soon 進(jìn)入 _ready 隊(duì)列，等待被執(zhí)行。
應(yīng)用場(chǎng)景：實(shí)戰(zhàn)項(xiàng)目中的避坑指南
回到實(shí)戰(zhàn)項(xiàng)目。假設(shè)你在做一個(gè)高并發(fā)的 WebSocket 聊天服務(wù)，用戶數(shù)上萬(wàn)。常見(jiàn)坑點(diǎn)有這三個(gè)：數(shù)據(jù)庫(kù)同步驅(qū)動(dòng)：asyncpg 是異步的，psycopg2 是同步的。如果你在 async 函數(shù)里用 psycopg2 查詢(xún)，哪怕只查 10ms，也會(huì)阻塞事件循環(huán) 10ms。如果有 1000 個(gè)并發(fā)，延遲就是 10 秒。解決方案：用 asyncio.to_thread 把同步調(diào)用扔到線程池，或者換用異步驅(qū)動(dòng)。任務(wù)未取消：用戶斷開(kāi)連接時(shí)，如果沒(méi)取消對(duì)應(yīng)的協(xié)程任務(wù)，它會(huì)繼續(xù)運(yùn)行，占用內(nèi)存。務(wù)必使用 try/except CancelledError 并清理資源。信號(hào)量濫用：用 asyncio.Semaphore 控制并發(fā)數(shù)是對(duì)的，但注意它的釋放必須在 finally 塊中。如果協(xié)程異常退出，信號(hào)量沒(méi)釋放，后續(xù)請(qǐng)求全部阻塞。這些坑，Stack Overflow 上每天都有人問(wèn)。但如果你懂 _run_once 的執(zhí)行流程，你就知道問(wèn)題出在哪。不要迷信框架，要理解底層。
你在項(xiàng)目里踩過(guò)這個(gè)坑嗎？評(píng)論區(qū)聊聊