Python asyncio異步編程中5大踩坑實錄
上周接了個需求，把一個同步爬蟲改成異步的。老板說「應該很快吧，不就是加幾個 async await 嘛」。我當時也覺得是，結(jié)果整整踩了三天坑，有兩天搞到凌晨一點多。今天把這些坑整理出來，希望后面的兄弟們少走點彎路。
先說結(jié)論
| 問題描述 | 嚴重程度 | 排查耗時 |
|---|---|---|
| 在 async 函數(shù)里調(diào)同步阻塞代碼 | ????? | 4h |
| 忘了 await 導致拿到協(xié)程對象 | ??? | 30min |
| aiohttp session 沒正確關閉 | ???? | 2h |
| 事件循環(huán)嵌套（loop 里套 loop） | ????? | 5h |
| 并發(fā)量沒控制導致被封 IP | ???? | 1h（加上等解封的時間就不止了） |
下面一個一個說。
坑 1：async 函數(shù)里混入了同步阻塞代碼
最致命的坑，因為它不報錯，只是慢。
我原來的代碼長這樣：
import asyncio
import requests # 注意這是同步庫
async def fetch_page(url: str) -> str:
# 看起來很正常對吧？但 requests.get 是同步阻塞的
resp = requests.get(url, timeout=10)
return resp.text
async def main():
urls = [f"https://httpbin.org/delay/1" for _ in range(10)]
tasks = [fetch_page(url) for url in urls]
results = await asyncio.gather(*tasks)
print(f"抓了 {len(results)} 個頁面")
asyncio.run(main())
跑完一看，10 個請求花了 10 秒多。不對啊，不是異步嗎，不應該 1 秒多就完事？
requests.get() 是同步阻塞調(diào)用。放在 async 函數(shù)里也一樣，執(zhí)行的時候還是會阻塞整個事件循環(huán)。asyncio 的事件循環(huán)是單線程的，一個任務阻塞了，其他任務全得等著。
給函數(shù)加 async 關鍵字不會讓里面的同步代碼變成異步的，這只是聲明了這個函數(shù)是個協(xié)程。
正確做法是換成 aiohttp：
import asyncio
import aiohttp
async def fetch_page(session: aiohttp.ClientSession, url: str) -> str:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
return await resp.text()
async def main():
urls = [f"https://httpbin.org/delay/1" for _ in range(10)]
async with aiohttp.ClientSession() as session:
tasks = [fetch_page(session, url) for url in urls]
results = await asyncio.gather(*tasks)
print(f"抓了 {len(results)} 個頁面")
asyncio.run(main())
這回 10 個請求 1.2 秒搞定。
如果實在沒法替換同步庫——比如某些數(shù)據(jù)庫驅(qū)動只有同步版本——可以用 asyncio.to_thread() 把同步調(diào)用丟到線程池里：
import asyncio
import requests
def sync_fetch(url: str) -> str:
"""這是個普通同步函數(shù)"""
resp = requests.get(url, timeout=10)
return resp.text
async def fetch_page(url: str) -> str:
# Python 3.9+ 可用，把同步函數(shù)丟到線程池執(zhí)行
return await asyncio.to_thread(sync_fetch, url)
async def main():
urls = [f"https://httpbin.org/delay/1" for _ in range(10)]
tasks = [fetch_page(url) for url in urls]
results = await asyncio.gather(*tasks)
print(f"抓了 {len(results)} 個頁面")
asyncio.run(main())
asyncio.to_thread 是 Python 3.9 加的，還在用 3.8 的話（該升了兄弟），用 loop.run_in_executor(None, sync_fetch, url) 也行。
坑 2：忘了 await，拿到一個協(xié)程對象
剛寫 asyncio 的時候真的很容易犯：
import asyncio
async def get_data():
await asyncio.sleep(1)
return {"status": "ok", "count": 42}
async def main():
data = get_data() # 忘了 await！
print(data) # <coroutine object get_data at 0x...>
print(data["status"]) # TypeError: 'coroutine' object is not subscriptable
asyncio.run(main())
控制臺還會給你一個 warning：RuntimeWarning: coroutine 'get_data' was never awaited。
這個 warning 其實挺明顯的，但日志多的時候，或者在 Jupyter 里跑，可能就淹沒了。
解決方案就是別忘了 await：
async def main():
data = await get_data() # 加上 await
print(data["status"]) # ok
我后來養(yǎng)成了一個習慣：凡是調(diào)用 async 函數(shù)，IDE 沒有高亮 await 關鍵字的，都多看一眼。用 PyCharm 或者 Cursor 的話，忘了 await 會有提示，這個功能真的能救命。
坑 3：aiohttp Session 沒正確關閉
這個坑比較隱蔽。代碼跑完會報一個 warning：
Unclosed client session client_session: <aiohttp.client.ClientSession object at 0x...>
我一開始的寫法：
import aiohttp
import asyncio
async def fetch(url: str) -> str:
session = aiohttp.ClientSession() # 每次調(diào)用都創(chuàng)建新 session
resp = await session.get(url)
text = await resp.text()
# 忘了關 session
return text
async def main():
urls = ["https://httpbin.org/get"] * 50
tasks = [fetch(url) for url in urls]
results = await asyncio.gather(*tasks)
print(f"完成 {len(results)} 個請求")
asyncio.run(main())
這段代碼有兩個問題。每次請求都創(chuàng)建新 Session：aiohttp 的 Session 內(nèi)部維護了連接池，頻繁創(chuàng)建銷毀等于放棄了連接復用，性能白白浪費。Session 沒關閉：會導致底層連接泄漏，請求量大了之后文件描述符耗盡，直接崩。
正確寫法：
import aiohttp
import asyncio
async def fetch(session: aiohttp.ClientSession, url: str) -> str:
async with session.get(url) as resp:
return await resp.text()
async def main():
urls = ["https://httpbin.org/get"] * 50
# 用 async with 確保 session 最終被關閉
async with aiohttp.ClientSession() as session:
tasks = [fetch(session, url) for url in urls]
results = await asyncio.gather(*tasks)
print(f"完成 {len(results)} 個請求")
asyncio.run(main())
一個 Session 搞定所有請求，用 async with 保證關閉。
坑 4：事件循環(huán)嵌套，這個真的折磨人
這個坑出現(xiàn)在我想在已有的 Flask 項目里調(diào)用 asyncio 代碼的時候。
import asyncio
async def async_work():
await asyncio.sleep(1)
return "done"
def sync_handler():
# 在同步代碼里調(diào)異步函數(shù)
result = asyncio.run(async_work()) # 第一次調(diào)沒問題
return result
# 但如果外層已經(jīng)有事件循環(huán)在跑（比如 Jupyter、某些框架內(nèi)部）：
# RuntimeError: asyncio.run() cannot be called from a running event loop
在 Jupyter Notebook 里這個問題 100% 必現(xiàn)，因為 Jupyter 自己就有一個事件循環(huán)在跑。
我試過幾種方案：
方案 A：nest_asyncio（快速解決，但不太優(yōu)雅）
import nest_asyncio
nest_asyncio.apply() # 允許事件循環(huán)嵌套
import asyncio
async def async_work():
await asyncio.sleep(1)
return "done"
# 現(xiàn)在 Jupyter 里也能用了
result = asyncio.run(async_work())
print(result)
這個庫就是打了個猴子補丁讓嵌套合法化，Jupyter 里用用可以，生產(chǎn)環(huán)境我不太敢。
方案 B：用線程跑獨立的事件循環(huán)（推薦）
import asyncio
from concurrent.futures import Future
import threading
def run_async_in_thread(coro):
"""在獨立線程中啟動新的事件循環(huán)來執(zhí)行協(xié)程"""
result_future: Future = Future()
def _run():
try:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
result = loop.run_until_complete(coro)
result_future.set_result(result)
except Exception as e:
result_future.set_exception(e)
finally:
loop.close()
thread = threading.Thread(target=_run)
thread.start()
thread.join()
return result_future.result()
async def async_work():
await asyncio.sleep(1)
return "done"
# 在同步代碼里安全調(diào)用異步函數(shù)
result = run_async_in_thread(async_work())
print(result) # done
這個方案在 Flask 項目里跑得挺穩(wěn)。當然如果項目可以全面切異步框架（FastAPI、Starlette），就沒這個問題了。我后來把那個 Flask 服務遷到 FastAPI 了，世界清凈了很多。
坑 5：并發(fā)量不控制，直接被封 IP
這個坑跟 asyncio 本身關系不大，但用了 asyncio 之后幾乎必然會遇到。
同步爬蟲天然就慢，很少觸發(fā)限流。換成異步以后，幾百個請求瞬間打出去，對面服務器直接把你封了。
import asyncio
import aiohttp
# 用信號量控制并發(fā)數(shù)
SEM = asyncio.Semaphore(10) # 最多 10 個并發(fā)
async def fetch(session: aiohttp.ClientSession, url: str) -> str:
async with SEM: # 獲取信號量，超過 10 個就等著
print(f"開始請求: {url}")
async with session.get(url) as resp:
text = await resp.text()
# 加個隨機延遲，別太暴力
await asyncio.sleep(0.5)
return text
async def main():
urls = [f"https://httpbin.org/get?page={i}" for i in range(100)]
connector = aiohttp.TCPConnector(limit=20) # 連接池也限制一下
async with aiohttp.ClientSession(connector=connector) as session:
tasks = [fetch(session, url) for url in urls]
results = await asyncio.gather(*tasks, return_exceptions=True)
success = sum(1 for r in results if not isinstance(r, Exception))
failed = sum(1 for r in results if isinstance(r, Exception))
print(f"成功: {success}, 失敗: {failed}")
asyncio.run(main())
幾個關鍵點：
asyncio.Semaphore：控制并發(fā)數(shù)的核心，比自己手寫隊列靠譜多了TCPConnector(limit=20)：限制底層 TCP 連接數(shù)return_exceptions=True：讓 gather 不會因為一個任務報錯就全部取消，失敗的任務會返回異常對象- 加延遲：
await asyncio.sleep()是異步的，不會阻塞別的任務。time.sleep()會阻塞整個循環(huán)——回到坑 1
額外說一個：異步代碼的異常處理
不算坑但容易忽略。asyncio.gather 默認行為是一個任務拋異常就取消其他所有任務：
import asyncio
async def good_task():
await asyncio.sleep(1)
return "我執(zhí)行完了"
async def bad_task():
await asyncio.sleep(0.5)
raise ValueError("我炸了")
async def main():
try:
# 默認行為：bad_task 一炸，good_task 也被取消
results = await asyncio.gather(good_task(), bad_task())
except ValueError as e:
print(f"捕獲到異常: {e}")
print("---")
# 加 return_exceptions=True：不會互相影響
results = await asyncio.gather(
good_task(), bad_task(), return_exceptions=True
)
for r in results:
if isinstance(r, Exception):
print(f"任務失敗: {r}")
else:
print(f"任務成功: {r}")
asyncio.run(main())
生產(chǎn)環(huán)境基本都要加 return_exceptions=True，不然一個請求失敗整批全廢，太虧了。
小結(jié)
回過頭來看，asyncio 的核心概念不復雜：事件循環(huán) + 協(xié)程 + await。但坑基本都出在異步和同步的邊界上：同步代碼混進異步函數(shù)會阻塞整個循環(huán)；同步環(huán)境調(diào)異步代碼會循環(huán)嵌套沖突；Session 忘關會泄漏；并發(fā)量不控制下游扛不住。
我個人的經(jīng)驗是，小項目別硬上 asyncio。只是寫個腳本抓十幾個頁面，多線程 + requests 完全夠用，代碼還好理解。asyncio 真正發(fā)揮威力的場景是高并發(fā) IO 密集型服務，比如 API 網(wǎng)關、WebSocket 服務、大批量數(shù)據(jù)采集。
還有就是，寫異步代碼之前先確認用到的所有庫都有異步版本。requests → aiohttp，psycopg2 → asyncpg，redis-py 現(xiàn)在自帶 async 支持了。如果核心依賴沒有異步版本，硬上 asyncio 意義不大，到處 to_thread 反而更亂。
到此這篇關于Python asyncio異步編程中5大踩坑實錄的文章就介紹到這了,更多相關Python asyncio異步編程內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
Python實現(xiàn)整數(shù)與浮點數(shù)相互轉(zhuǎn)換的方法
在Python編程的浩瀚宇宙中,數(shù)字類型是最基礎卻最強大的基石之一,整數(shù)（int）和浮點數(shù)（float）作為日常開發(fā)中不可或缺的兩種數(shù)據(jù)類型,它們之間的相互轉(zhuǎn)換看似簡單,卻暗藏玄機,本文將帶你深入探索Python中整數(shù)與浮點數(shù)相互轉(zhuǎn)換的完整知識體系,需要的朋友可以參考下2026-04-04
這篇文章主要為大家詳細介紹了python八皇后問題的解決方法，具有一定的參考價值，感興趣的小伙伴們可以參考一下2018-09-09
在當今數(shù)字化時代,情感分析與情緒識別技術(shù)日益成為人機交互,社交媒體分析,智能客服等領域的重要應用,下面我們就來深入聊聊基于Python的情感分析與情緒識別技術(shù)吧2025-03-03
這篇文章主要介紹了Python中Anaconda3 安裝gdal庫的方法,本文給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2021-04-04
這篇文章主要介紹了Python根據(jù)服務獲取端口號,文中給大家提到了linux查看端口開啟端口的方法，需要的朋友可以參考下2019-09-09
python Tornado事件循環(huán)示例源碼解析
這篇文章主要為大家介紹了python Tornado事件循環(huán)示例源碼解析,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進步,早日升職加薪2023-09-09
這篇文章主要介紹了Python中的集合類型知識講解,是Python入門學習中的基礎知識,需要的朋友可以參考下2015-08-08
使用基于Python的Tornado框架的HTTP客戶端的教程
這篇文章主要介紹了制作一個基于Python的Tornado框架的HTTP客戶端的教程,Tornado的異步特性使其能夠獲得很好的性能,需要的朋友可以參考下2015-04-04
解決python環(huán)境無法安裝 exceptions的問題(含錯誤原因分析)
pip install exceptions失敗的根本原因是?exceptions模塊在Python 3.x中已被移除,因此無法通過pip安裝,本文給大家介紹解決python環(huán)境無法安裝exceptions的問題,感興趣的朋友跟隨小編一起看看吧2025-09-09
這篇文章主要介紹了python實現(xiàn)的文件同步服務器,實例分析了文件同步服務器的原理及客戶端、服務端的實現(xiàn)技巧,需要的朋友可以參考下2015-06-06