Python异步编程：async/await与asyncio实战
写完Deep Agents那几篇，后台不少老铁私信问代码里的async、await到底啥意思。翻了一圈网上的教程，要么是简单的Hello World，要么直接甩官方文档让你自己啃。
今天直接上硬货，把异步编程掰开揉碎讲明白，配上能直接跑起来的代码和真实应用场景。
一、先搞懂概念：异步到底是个啥
传统的同步代码就像去银行办业务——排队，一个人办完才到下一个人。如果前面的人办得慢，后面的人只能干等。
异步代码则像去奶茶店——你点完单拿个号，先去逛别的店，等奶茶做好了手机会提醒你回来拿。
异步编程的核心：让程序在等待I/O操作（网络请求、文件读写、数据库查询）的时候，不傻等，而是去干别的活。
Python实现异步靠三个关键词：
- async def：定义一个异步函数（官方叫协程）
- await：在异步函数里等另一个异步操作完成，同时让出CPU
- asyncio：Python自带的异步运行时，负责调度这些任务
二、最基础的例子：别被语法吓到
2.1 定义一个异步函数
import asyncio
async def say_hello():
print("Hello")
await asyncio.sleep(1) # 模拟耗时操作
print("World")
# 跑起来
asyncio.run(say_hello())
输出：
Hello
World
注意：asyncio.sleep(1)和time.sleep(1)不一样。前者是异步睡眠，睡眠期间事件循环可以去干别的；后者是同步阻塞，整个线程都卡住。
2.2 直接调用异步函数？不行
新手最容易踩的坑：
async def foo():
return 42
# 这不会执行函数，只是创建一个协程对象
result = foo()
print(result) # 输出：<coroutine object foo at 0x…>
# 正确姿势
result = await foo() # 必须在async函数里用await
# 或者
result = asyncio.run(foo()) # 顶层入口用这个
官方文档明确说：单纯调用异步函数不会调度执行，必须用await或asyncio.create_task 。
三、并发执行：这才是异步的灵魂
3.1 用asyncio.gather同时跑多个任务
先看同步版本：
import time
def task(name, delay):
print(f"{name} 开始")
time.sleep(delay) # 同步阻塞
print(f"{name} 结束")
return f"{name}的结果"
start = time.time()
result1 = task("任务1", 3)
result2 = task("任务2", 2)
result3 = task("任务3", 1)
print(f"总耗时：{time.time() – start:.2f}秒")
输出（总耗时约6秒）：
任务1 开始
任务1 结束
任务2 开始
任务2 结束
任务3 开始
任务3 结束
总耗时：6.00秒
再看异步版本：
import asyncio
import time
async def task(name, delay):
print(f"{name} 开始")
await asyncio.sleep(delay) # 异步阻塞
print(f"{name} 结束")
return f"{name}的结果"
async def main():
start = time.time()
# 并发执行三个任务
results = await asyncio.gather(
task("任务1", 3),
task("任务2", 2),
task("任务3", 1)
)
print(f"结果：{results}")
print(f"总耗时：{time.time() – start:.2f}秒")
asyncio.run(main())
输出（总耗时约3秒）：
任务1 开始
任务2 开始
任务3 开始
任务3 结束
任务2 结束
任务1 结束
结果：['任务1的结果', '任务2的结果', '任务3的结果']
总耗时：3.00秒
看明白了吗？三个任务同时开始，耗时最长的任务3秒决定了总时间。这就是异步的威力。
3.2 用create_task创建后台任务
有时候不想等任务完成，让它后台跑着就行：
import asyncio
async def background_task():
print("后台任务开始")
await asyncio.sleep(5)
print("后台任务结束")
async def main():
# 创建后台任务
task = asyncio.create_task(background_task())
print("主程序继续干别的")
await asyncio.sleep(2)
print("主程序干完了")
# 如果需要等后台任务，可以await
# await task
asyncio.run(main())
输出：
后台任务开始
主程序继续干别的
主程序干完了
后台任务结束
注意：如果不保存任务的引用，任务可能被垃圾回收直接干掉。官方推荐用集合保存引用：
background_tasks = set()
for i in range(10):
task = asyncio.create_task(some_coro())
background_tasks.add(task)
# 任务完成自动从集合移除
task.add_done_callback(background_tasks.discard)
3.3 Python 3.11+ 的新姿势：TaskGroup
Python 3.11引入了TaskGroup，用上下文管理器管理任务生命周期：
async def main():
async with asyncio.TaskGroup() as tg:
task1 = tg.create_task(task("任务1", 3))
task2 = tg.create_task(task("任务2", 2))
task3 = tg.create_task(task("任务3", 1))
# 退出with块时自动等待所有任务完成
print("所有任务都完成了")
好处是如果某个任务挂了，会自动取消其他任务，避免僵尸任务。
四、超时控制：别让任务跑太久
网络请求最怕卡死，asyncio.wait_for可以设置超时：
import asyncio
async def slow_operation():
await asyncio.sleep(10)
return "终于完成了"
async def main():
try:
# 最多等3秒
result = await asyncio.wait_for(slow_operation(), timeout=3)
print(result)
except asyncio.TimeoutError:
print("操作超时了，不等了")
asyncio.run(main())
输出：
操作超时了，不等了
超时后任务会被自动取消，抛CancelledError。如果你的任务有清理逻辑，记得用try/finally捕获。
五、异常处理：异步任务挂了怎么办
5.1 单个任务的异常
async def might_fail():
await asyncio.sleep(1)
raise ValueError("出错了")
async def main():
try:
await might_fail()
except ValueError as e:
print(f"捕获到异常：{e}")
asyncio.run(main())
5.2 gather任务的异常
gather默认只要有一个任务抛异常，整体就抛异常：
async def main():
try:
results = await asyncio.gather(
might_fail(),
task("正常任务", 2)
)
except ValueError as e:
print(f"捕获到：{e}")
如果想收集所有结果（包括异常），用return_exceptions=True：
results = await asyncio.gather(
might_fail(),
task("正常任务", 2),
return_exceptions=True
)
# results = [ValueError('出错了'), '正常任务的结果']
5.3 TaskGroup的异常处理
TaskGroup会把所有异常打包成ExceptionGroup抛出：
async def main():
try:
async with asyncio.TaskGroup() as tg:
tg.create_task(might_fail())
tg.create_task(task("正常", 2))
except* ValueError as eg:
for e in eg.exceptions:
print(f"处理ValueError: {e}")
六、实战场景：这玩意儿到底用在哪
6.1 场景一：批量网络请求（性能提升10倍）
这是异步最典型的应用。有开发者实测：下载10次10MB的文件，同步耗时约12秒，异步仅需1.3秒。
import asyncio
import aiohttp # 第三方异步HTTP库
import time
async def fetch_url(session, url):
async with session.get(url) as response:
return await response.text()
async def fetch_all(urls):
async with aiohttp.ClientSession() as session:
tasks = [fetch_url(session, url) for url in urls]
return await asyncio.gather(*tasks)
async def main():
urls = ["http://example.com"] * 10
start = time.time()
results = await fetch_all(urls)
print(f"抓取{len(urls)}个页面，耗时：{time.time()–start:.2f}秒")
asyncio.run(main())
6.2 场景二：Web API服务
用FastAPI或Sanic这类异步框架，能同时处理成千上万的长连接。
from fastapi import FastAPI
import asyncio
import httpx # 支持异步的HTTP客户端
app = FastAPI()
@app.get("/proxy/{url}")
async def proxy(url: str):
# 这个请求不会阻塞其他请求
async with httpx.AsyncClient() as client:
resp = await client.get(f"http://{url}")
return {"content": resp.text[:100]}
6.3 场景三：TCP长连接服务
用asyncio.open_connection写个简单的echo服务器：
import asyncio
async def handle_client(reader, writer):
addr = writer.get_extra_info('peername')
print(f"新连接：{addr}")
while True:
data = await reader.read(100)
if not data:
break
message = data.decode()
print(f"收到：{message}")
writer.write(data) # 原样返回
await writer.drain()
print(f"关闭连接：{addr}")
writer.close()
await writer.wait_closed()
async def main():
server = await asyncio.start_server(
handle_client, '127.0.0.1', 8888
)
async with server:
await server.serve_forever()
asyncio.run(main())
6.4 场景四：异步数据库操作
用asyncpg操作PostgreSQL，aiomysql操作MySQL，motor操作MongoDB。下面是个简单的例子：
import asyncio
import asyncpg
async def main():
# 连接数据库（异步）
conn = await asyncpg.connect(
user='user', password='pass',
database='test', host='localhost'
)
# 执行查询（不会阻塞）
rows = await conn.fetch('SELECT * FROM users')
for row in rows:
print(row)
await conn.close()
asyncio.run(main())
七、避坑指南：老司机踩过的雷
7.1 不要在异步函数里用time.sleep()
async def bad():
time.sleep(5) # ❌ 整个事件循环卡死5秒
await asyncio.sleep(5) # ✅ 正确姿势
7.2 不要搞混并发和并行
异步是并发（concurrency），不是并行（parallelism）。它在一个线程里来回切换任务，适合I/O密集型；CPU密集型任务（比如视频编码）用多进程。
7.3 记得加超时
网络请求永远要设超时，不然被慢服务拖死：
try:
result = await asyncio.wait_for(api_call(), timeout=5.0)
except asyncio.TimeoutError:
# 走降级逻辑
result = fallback_data
7.4 保存Task引用
前面说过了，不保存引用任务可能被GC回收：
# ❌ 危险写法
asyncio.create_task(background_job())
# ✅ 安全写法
task = asyncio.create_task(background_job())
# 或者在集合里存着
八、总结：什么时候该用异步
适合异步的场景：
- 网络爬虫（大量HTTP请求）
- Web服务（处理大量并发连接）
- API网关（转发请求到后端）
- 消息队列消费者
- 数据库连接池
不适合异步的场景：
- CPU密集型计算（图像处理、视频编码、机器学习训练）
- 简单脚本（杀鸡不用牛刀）
- 已有同步代码库（改造成本太高）
最后送大家一句话：异步不是银弹，但在I/O密集型场景下，是真香。
有问题评论区见。
