由淺入深介紹python asyncio的各種用法與代碼示例
更新時間：2025年11月05日 09:57:27 作者：令狐掌門
asyncio 是 Python 的一個庫,用于編寫并發(fā)代碼,使用協(xié)程、任務(wù)和 Futures 來處理 I/O 密集型和高延遲操作,下面小編就為大家由淺入深介紹python asyncio的各種用法吧
下面是一份“由淺入深”的 asyncio 實(shí)戰(zhàn)手冊。先解釋核心概念，再給出可直接運(yùn)行的、循序漸進(jìn)的示例代碼。全部示例都只依賴標(biāo)準(zhǔn)庫（除少數(shù)標(biāo)注處），適配 Python 3.10+（如使用 TaskGroup、asyncio.timeout() 的段落需要 3.11+）。
1. 基礎(chǔ)概念速覽
async def：定義協(xié)程函數(shù)（coroutine function）。await：在協(xié)程中等待另一個可等待對象（協(xié)程、asyncio.Task、asyncio.Future等）。- 事件循環(huán)（Event Loop）：調(diào)度協(xié)程、I/O 事件的核心；用
asyncio.run(main())啟動。 - 并發(fā) ≠ 并行：asyncio 是單線程協(xié)作式并發(fā)，靠 I/O 等待時讓出控制權(quán)。
- 任務(wù)（Task）：把協(xié)程“提交給”事件循環(huán)讓其并發(fā)執(zhí)行：
asyncio.create_task(coro())。
2. 最小可運(yùn)行示例：async/await+asyncio.run
# demo_basic.py
import asyncio
async def io_job(name, delay):
print(f"[{name}] start")
await asyncio.sleep(delay) # 模擬I/O等待
print(f"[{name}] done after {delay}s")
return name, delay
async def main():
r1 = await io_job("A", 1)
r2 = await io_job("B", 2) # 串行等待，總耗時約3秒
print("results:", r1, r2)
if __name__ == "__main__":
asyncio.run(main())
3. 并發(fā)執(zhí)行：create_task、gather、as_completed
# demo_concurrency.py
import asyncio, random
async def fetch(i):
d = random.uniform(0.5, 2.0)
await asyncio.sleep(d)
return f"task-{i}", round(d, 2)
async def main():
# 3.1 create_task + await
t1 = asyncio.create_task(fetch(1))
t2 = asyncio.create_task(fetch(2))
r1 = await t1
r2 = await t2
print("create_task results:", r1, r2)
# 3.2 gather（順序返回結(jié)果；遇異常默認(rèn)會立刻拋出）
results = await asyncio.gather(*(fetch(i) for i in range(3, 8)))
print("gather results:", results)
# 3.3 as_completed（誰先完成先拿誰）
tasks = [asyncio.create_task(fetch(i)) for i in range(8, 13)]
for fut in asyncio.as_completed(tasks):
print("as_completed:", await fut)
if __name__ == "__main__":
asyncio.run(main())
4. 超時控制與取消：wait_for、asyncio.timeout、Task.cancel
# demo_timeout_cancel.py
import asyncio
async def slow():
try:
await asyncio.sleep(5)
return "ok"
except asyncio.CancelledError:
print("slow() got cancelled!")
raise
async def main():
# 4.1 wait_for（3.8+）
try:
res = await asyncio.wait_for(slow(), timeout=2)
print("result:", res)
except asyncio.TimeoutError:
print("wait_for timeout!")
# 4.2 asyncio.timeout 上下文（3.11+）
try:
async with asyncio.timeout(2): # 注：需 Python 3.11+
await slow()
except TimeoutError:
print("context timeout!")
# 4.3 手動取消
t = asyncio.create_task(slow())
await asyncio.sleep(1)
t.cancel()
try:
await t
except asyncio.CancelledError:
print("task cancelled confirmed")
if __name__ == "__main__":
asyncio.run(main())
小貼士：被取消的任務(wù)應(yīng)正確處理 CancelledError，并在需要時做清理（finally）。
5. 限流與同步原語：Semaphore、Lock、Event、Condition
# demo_sync_primitives.py
import asyncio, random
sem = asyncio.Semaphore(3) # 同時最多3個并發(fā)
lock = asyncio.Lock()
evt = asyncio.Event()
async def worker(i):
async with sem: # 限制并發(fā)
await asyncio.sleep(random.uniform(0.2, 1.0))
async with lock: # 保護(hù)共享輸出（示意）
print(f"worker {i} done")
async def notifier():
await asyncio.sleep(1)
evt.set() # 廣播事件
async def waiter():
print("waiting for event...")
await evt.wait()
print("event received!")
async def main():
tasks = [asyncio.create_task(worker(i)) for i in range(10)]
tasks += [asyncio.create_task(notifier()), asyncio.create_task(waiter())]
await asyncio.gather(*tasks)
if __name__ == "__main__":
asyncio.run(main())
Condition 適合更復(fù)雜的“等待某條件成立”的場景，用法與 threading.Condition 類似（只是換成 async with / await）。
6. 生產(chǎn)者-消費(fèi)者：asyncio.Queue
# demo_queue.py
import asyncio, random
async def producer(q: asyncio.Queue):
for i in range(10):
await asyncio.sleep(random.uniform(0.1, 0.4))
await q.put((i, f"data-{i}"))
print(f"produced {i}")
await q.put(None) # 結(jié)束哨兵
async def consumer(q: asyncio.Queue):
while True:
item = await q.get()
if item is None:
q.task_done()
break
i, data = item
await asyncio.sleep(0.3)
print(f"consumed {i} -> {data}")
q.task_done()
async def main():
q = asyncio.Queue(maxsize=5)
prod = asyncio.create_task(producer(q))
cons = asyncio.create_task(consumer(q))
await asyncio.gather(prod)
await q.join() # 等全部消費(fèi)完成
await cons # 等消費(fèi)者退出
if __name__ == "__main__":
asyncio.run(main())
7. 任務(wù)編組（Python 3.11+）：asyncio.TaskGroup
# demo_taskgroup.py
import asyncio, random
async def job(n):
await asyncio.sleep(random.uniform(0.2, 1.0))
if n == 3:
raise RuntimeError("boom at 3")
return n
async def main():
try:
async with asyncio.TaskGroup() as tg:
tasks = [tg.create_task(job(i)) for i in range(5)]
# 出錯會自動取消其余任務(wù)并向外傳播異常
except* RuntimeError as eg: # PEP 654（ExceptionGroup）
print("caught:", eg)
if __name__ == "__main__":
asyncio.run(main())
對比 gather：TaskGroup 在結(jié)構(gòu)化并發(fā)上更可靠，失敗會自動收攏與傳播。
8. 與線程/同步代碼協(xié)作：to_thread、run_in_executor
# demo_thread_bridge.py
import asyncio, time, concurrent.futures
def blocking_io(n):
time.sleep(n)
return f"blocking {n}s"
async def main():
# 8.1 Python 3.9+ 推薦：to_thread
r1 = await asyncio.to_thread(blocking_io, 1)
print("to_thread:", r1)
# 8.2 傳統(tǒng)：run_in_executor
loop = asyncio.get_running_loop()
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as pool:
futs = [loop.run_in_executor(pool, blocking_io, i) for i in (1, 2, 1)]
for r in await asyncio.gather(*futs):
print("executor:", r)
if __name__ == "__main__":
asyncio.run(main())
原則：CPU 密集就放線程/進(jìn)程池；I/O 密集用 await 原生異步接口。
9. TCP/UDP 網(wǎng)絡(luò)編程（內(nèi)置 Streams / Protocols）
9.1 TCP Echo（Server & Client，基于 Streams）
# demo_tcp_echo.py
import asyncio
async def handle_echo(reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
addr = writer.get_extra_info('peername')
print(f"client connected: {addr}")
try:
while data := await reader.readline():
msg = data.decode().rstrip()
print(f"recv: {msg}")
writer.write((msg + "\n").encode())
await writer.drain()
except asyncio.CancelledError:
raise
finally:
writer.close()
await writer.wait_closed()
print("client closed", addr)
async def run_server():
server = await asyncio.start_server(handle_echo, "127.0.0.1", 8888)
addrs = ", ".join(str(sock.getsockname()) for sock in server.sockets)
print(f"Serving on {addrs}")
async with server:
await server.serve_forever()
async def run_client():
reader, writer = await asyncio.open_connection("127.0.0.1", 8888)
for i in range(3):
writer.write(f"hello {i}\n".encode())
await writer.drain()
echo = await reader.readline()
print("echo:", echo.decode().rstrip())
writer.close()
await writer.wait_closed()
async def main():
server_task = asyncio.create_task(run_server())
await asyncio.sleep(0.2)
await run_client()
server_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await server_task
if __name__ == "__main__":
import contextlib
asyncio.run(main())
9.2 UDP（Datagram）
# demo_udp.py
import asyncio
class EchoServer(asyncio.DatagramProtocol):
def datagram_received(self, data, addr):
print("server recv:", data, "from", addr)
self.transport.sendto(data, addr)
async def main():
loop = asyncio.get_running_loop()
transport, _ = await loop.create_datagram_endpoint(
lambda: EchoServer(), local_addr=("127.0.0.1", 9999)
)
# client
on_resp = loop.create_future()
class Client(asyncio.DatagramProtocol):
def datagram_received(self, data, addr):
print("client recv:", data)
on_resp.set_result(None)
ctransport, _ = await loop.create_datagram_endpoint(
lambda: Client(), remote_addr=("127.0.0.1", 9999)
)
ctransport.sendto(b"hello-udp")
await on_resp
transport.close()
ctransport.close()
if __name__ == "__main__":
asyncio.run(main())
10. 子進(jìn)程（異步等待）：asyncio.create_subprocess_exec
# demo_subprocess.py
import asyncio, sys
async def main():
# 跨平臺示例：調(diào)用 python -c 'print("hi")'
proc = await asyncio.create_subprocess_exec(
sys.executable, "-c", 'print("hi from child")',
stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
)
out, err = await proc.communicate()
print("stdout:", out.decode().strip(), "| code:", proc.returncode)
if __name__ == "__main__":
asyncio.run(main())
11. 異步上下文管理器與迭代器：async with、async for
# demo_async_with_for.py
import asyncio
from contextlib import asynccontextmanager
@asynccontextmanager
async def open_resource():
print("acquire resource")
await asyncio.sleep(0.2)
try:
yield "RESOURCE"
finally:
await asyncio.sleep(0.2)
print("release resource")
class AsyncCounter:
def __init__(self, n): self.n=n; self.i=0
def __aiter__(self): return self
async def __anext__(self):
if self.i >= self.n:
raise StopAsyncIteration
await asyncio.sleep(0.1)
self.i += 1
return self.i
async def main():
async with open_resource() as r:
print("using:", r)
async for x in AsyncCounter(5):
print("got:", x)
if __name__ == "__main__":
asyncio.run(main())
12. 超實(shí)用模式集
12.1 背壓與批處理
# demo_backpressure.py
import asyncio, random
async def producer(q):
for i in range(30):
await q.put(i) # maxsize 限制可形成背壓
await asyncio.sleep(0.05)
await q.put(None)
async def consumer(q):
batch = []
while True:
item = await q.get()
if item is None:
if batch:
print("flush batch:", batch)
q.task_done()
break
batch.append(item)
if len(batch) >= 8:
# 模擬批量處理
await asyncio.sleep(random.uniform(0.1, 0.3))
print("process batch:", batch)
batch.clear()
q.task_done()
async def main():
q = asyncio.Queue(maxsize=10)
await asyncio.gather(producer(q), consumer(q))
await q.join()
if __name__ == "__main__":
asyncio.run(main())
12.2 冪等重試 + 指數(shù)退避
# demo_retry.py
import asyncio, random
async def fragile_call():
await asyncio.sleep(0.1)
if random.random() < 0.7:
raise RuntimeError("transient")
return "ok"
async def retry(coro_func, attempts=5, base=0.2):
for n in range(attempts):
try:
return await coro_func()
except Exception as e:
if n == attempts - 1:
raise
await asyncio.sleep(base * (2 ** n)) # 退避
raise RuntimeError("unreachable")
async def main():
try:
r = await retry(fragile_call)
print("result:", r)
except Exception as e:
print("failed:", e)
if __name__ == "__main__":
asyncio.run(main())
13. 常見坑與最佳實(shí)踐
入口統(tǒng)一用 asyncio.run(main())，不要混用早期 API（如手動獲取 loop、run_until_complete），除非有特殊需求。
避免阻塞調(diào)用（如 time.sleep()、重 CPU 任務(wù)）直接出現(xiàn)在協(xié)程里；改用 await asyncio.sleep() 或 asyncio.to_thread()/進(jìn)程池。
正確處理取消：在 try/except/finally 中傳播 CancelledError，避免吞掉取消導(dǎo)致任務(wù)“僵尸化”。
使用限流：對外部服務(wù)/磁盤/網(wǎng)絡(luò)做 Semaphore、隊列背壓，保護(hù)系統(tǒng)。
結(jié)構(gòu)化并發(fā)：優(yōu)先考慮 TaskGroup（3.11+）來讓“任務(wù)生命周期”明確，異常聚合安全。
日志與超時：給關(guān)鍵 I/O 加 timeout，并使用 asyncio.create_task() 后保存句柄，便于監(jiān)控和取消。
14. 進(jìn)階延伸（可選）
文件異步：標(biāo)準(zhǔn)庫沒有真正異步文件 I/O；可選三方庫 aiofiles（僅示意）：
# pip install aiofiles
import aiofiles, asyncio
async def read_file(p):
async with aiofiles.open(p, "r", encoding="utf-8") as f:
return await f.read()
HTTP 客戶端/服務(wù)端：aiohttp、httpx[http2] 等第三方庫更貼近真實(shí)網(wǎng)絡(luò)場景。
與 GUI/Qt 的集成：可通過 qasync 把 Qt 事件循環(huán)與 asyncio 融合（適合你在 QML/PySide6 下需要異步網(wǎng)絡(luò)/IO 的情況）。
到此這篇關(guān)于由淺入深介紹python asyncio的各種用法與代碼示例的文章就介紹到這了,更多相關(guān)python asyncio用法內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
在本篇文章里小編給大家整理的是一篇關(guān)于python常見的占位符總結(jié)及用法，有興趣的朋友們可以跟著學(xué)習(xí)參考下。2021-07-07
Python OpenCV調(diào)用攝像頭檢測人臉并截圖
這篇文章主要為大家詳細(xì)介紹了Python OpenCV調(diào)用攝像頭檢測人臉并截圖，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價值，感興趣的小伙伴們可以參考一下2019-07-07
統(tǒng)計檢驗(yàn)是數(shù)據(jù)分析中的重要工具,用于檢驗(yàn)數(shù)據(jù)集中的差異、關(guān)聯(lián)和分布等統(tǒng)計性質(zhì),本文為大家整理了常見的統(tǒng)計檢驗(yàn)方法,希望對大家有所幫助2024-01-01
如何利用python寫GUI及生成.exe可執(zhí)行文件
工作中需要開發(fā)一個小工具,簡單的UI界面可以很好的提高工具的實(shí)用性,由此開啟了我的第一次GUI開發(fā)之旅,這篇文章主要給大家介紹了關(guān)于如何利用python寫GUI及生成.exe可執(zhí)行文件的相關(guān)資料,需要的朋友可以參考下2021-12-12
Python數(shù)據(jù)類型之Tuple元組實(shí)例詳解
這篇文章主要介紹了Python數(shù)據(jù)類型之Tuple元組,結(jié)合實(shí)例形式分析了Python元組類型的概念、定義、讀取、連接、判斷等常見操作技巧與相關(guān)注意事項,需要的朋友可以參考下2019-05-05
這篇文章主要介紹了python與json數(shù)據(jù)的交互詳情，json是一種獨(dú)立于編程語言和平臺的輕量級數(shù)據(jù)交換方式，更多相關(guān)內(nèi)容介紹，需要的朋友可以參考一下2022-07-07
這篇文章主要為大家詳細(xì)介紹了使用python實(shí)現(xiàn)ANN的相關(guān)資料，具有一定的參考價值，感興趣的小伙伴們可以參考一下2017-12-12
Python中不同進(jìn)制的語法及轉(zhuǎn)換方法分析
這篇文章主要介紹了Python中不同進(jìn)制的語法及轉(zhuǎn)換方法,結(jié)合實(shí)例形式分析了Python不同進(jìn)制的表示方法及相互轉(zhuǎn)換方法,需要的朋友可以參考下2016-07-07
Python+eval函數(shù)實(shí)現(xiàn)動態(tài)地計算數(shù)學(xué)表達(dá)式詳解
Python的 eval() 允許從基于字符串或基于編譯代碼的輸入中計算任意Python表達(dá)式。當(dāng)從字符串或編譯后的代碼對象的任何輸入中動態(tài)計算Python表達(dá)式時，此函數(shù)非常方便。本文將利用eval實(shí)現(xiàn)動態(tài)地計算數(shù)學(xué)表達(dá)式，需要的可以參考一下2022-09-09
Python通過四大 AutoEDA 工具包快速產(chǎn)出完美數(shù)據(jù)報告
在三年前，我們做數(shù)據(jù)競賽或者數(shù)據(jù)建模類的項目時，前期我們會耗費(fèi)較多的時間去分析數(shù)據(jù)，但現(xiàn)在非常多擅長數(shù)據(jù)分析的大師們已經(jīng)將我們平時?？吹臄?shù)據(jù)方式進(jìn)行了集成，開發(fā)了很多AutoEDA的工具包?？梢詭椭覀児?jié)省大量時間2021-11-11