Python的异步编程很多人觉得难懂，其实核心概念就几个。本文从最基础的回调讲起，一步步到现代async/await写法，帮你彻底理解Python异步。

为什么需要异步？

同步代码在等IO时线程会阻塞，异步则可以在等待时去做别的事：

import requests
resp = requests.get("https://httpbin.org/delay/3") # 阻塞3秒
print(resp.status_code)

# 异步：等的同时可以做其他事
import asyncio
import aiohttp

async def fetch():
async with aiohttp.ClientSession() as session:
async with session.get("https://httpbin.org/delay/3") as resp:
return await resp.json()

# 同时发5个请求，总耗时还是3秒而非15秒
results = await asyncio.gather(*[fetch() for _ in range(5)])

事件循环：异步的心脏

事件循环（Event Loop）就是一个不停检查"谁准备好了"的调度器：

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

# 事件循环的工作：检查任务 → 执行到await → 切到另一个任务 → 重复
loop.run_until_complete(my_async_function())

Python 3.10+ 推荐用 asyncio.run()：

print("hello")

asyncio.run(main()) # 自动创建循环并运行

async/await语法

async def 定义协程，await 等待异步操作完成：

async def cook_pasta():
print("开始煮面...")
await asyncio.sleep(3) # 模拟煮面3分钟（不阻塞）
print("面煮好了！")
return "意大利面"

async def make_sauce():
print("开始做酱...")
await asyncio.sleep(2) # 模拟做酱2分钟
print("酱做好了！")
return "番茄酱"

async def dinner():
# 同时煮面和做酱，总耗时3分钟而非5分钟
pasta, sauce = await asyncio.gather(cook_pasta(), make_sauce())
print(f"晚餐：{pasta}配{sauce}")

asyncio.run(dinner())

输出：

开始做酱...
酱做好了！
面煮好了！
晚餐：意大利面配番茄酱

实战：异步HTTP请求

2026年最常用的异步HTTP库是 aiohttp 和 httpx：

import httpx

async def fetch_url(client, url):
try:
resp = await client.get(url, timeout=10.0)
print(f"{url} → {resp.status_code}")
return resp.text[:100]
except Exception as e:
print(f"{url} → 错误: {e}")
return None

async def main():
urls = [
"https://httpbin.org/get",
"https://httpbin.org/ip",
"https://httpbin.org/headers",
"https://httpbin.org/user-agent",
]

async with httpx.AsyncClient() as client:
results = await asyncio.gather(
*[fetch_url(client, url) for url in urls]
)

for url, result in zip(urls, results):
if result:
print(f"{url}: {result[:50]}...")

asyncio.run(main())

常见陷阱

1. 忘记await

result = asyncio.sleep(1)

# ✅ 正确
result = await asyncio.sleep(1)

2. 在异步函数里用同步IO

import requests
resp = requests.get("https://example.com")

# ✅ 用异步库
import httpx
async with httpx.AsyncClient() as client:
resp = await client.get("https://example.com")

3. asyncio.gather不处理异常

results = await asyncio.gather(task1(), task2())

# ✅ return_exceptions=True
results = await asyncio.gather(task1(), task2(), return_exceptions=True)

异步 vs 多线程

维度	async/await	多线程
并发模型	协作式	抢占式
适合场景	IO密集（网络/文件）	CPU密集
调试难度	较低	较高（竞态条件）
GIL影响	无	有
上下文切换成本	极低	较高

简单规则：主要等网络/磁盘 → 用异步；主要做计算 → 用多线程/多进程。

进阶：异步上下文管理器和迭代器

class AsyncDB:
async def __aenter__(self):
self.conn = await connect_db()
return self

async def __aexit__(self, *args):
await self.conn.close()

async def query(self, sql):
return await self.conn.execute(sql)

# 使用
async with AsyncDB() as db:
result = await db.query("SELECT 1")

# 异步迭代器
async def read_lines(path):
async with aiofiles.open(path) as f:
async for line in f:
yield line.strip()

async for line in read_lines("big_file.txt"):
print(line)

Python异步的核心就是事件循环 + 协程。理解了这两个概念，剩下的就是API的使用了。建议从httpx异步请求开始练手，这是最快感受到异步优势的场景。