前言
Python 的 asyncio 是编写并发代码的标准库，使用 async/await 语法实现单线程下的高并发 IO 操作。本文将从协程基础讲到实际应用，帮你掌握 Python 异步编程。
一、同步 vs 异步
1.1 同步代码的问题
import requests
import time
def fetch_url(url):
resp = requests.get(url)
return resp.status_code
start = time.time()
for url in ["https://httpbin.org/delay/1"] * 10:
fetch_url(url)
print(f"同步耗时: {time.time() - start:.2f}s") # ~10s10 个请求串行执行，每个等待 1 秒，总共 10 秒。
1.2 异步的优势
import asyncio
import aiohttp
import time
async def fetch_url(session, url):
async with session.get(url) as resp:
return resp.status
async def main():
async with aiohttp.ClientSession() as session:
tasks = [fetch_url(session, "https://httpbin.org/delay/1") for _ in range(10)]
results = await asyncio.gather(*tasks)
print(results)
start = time.time()
asyncio.run(main())
print(f"异步耗时: {time.time() - start:.2f}s") # ~1s10 个请求并发执行，总共约 1 秒。
二、协程基础
2.1 定义和调用协程
import asyncio
async def hello():
print("Hello")
await asyncio.sleep(1)
print("World")
asyncio.run(hello())2.2 await 的含义
await 只能在 async 函数内使用，它会：
- 暂停当前协程的执行
- 将控制权交还给事件循环
- 等待 awaitable 对象完成后恢复执行
async def task_a():
print("A start")
await asyncio.sleep(1)
print("A end")
async def task_b():
print("B start")
await asyncio.sleep(1)
print("B end")
async def main():
await task_a()
await task_b()
asyncio.run(main())三、并发执行
3.1 asyncio.gather
async def main():
await asyncio.gather(task_a(), task_b())3.2 asyncio.create_task
async def main():
task1 = asyncio.create_task(task_a())
task2 = asyncio.create_task(task_b())
print("doing other work")
await task1
await task23.3 超时控制
async def slow_operation():
await asyncio.sleep(10)
return "done"
async def main():
try:
result = await asyncio.wait_for(slow_operation(), timeout=3.0)
print(result)
except asyncio.TimeoutError:
print("操作超时!")
asyncio.run(main())3.4 asyncio.as_completed
async def fetch(simulated_time, name):
await asyncio.sleep(simulated_time)
return f"{name}: {simulated_time}s"
async def main():
tasks = [
fetch(3, "task1"),
fetch(1, "task2"),
fetch(2, "task3"),
]
for coro in asyncio.as_completed(tasks):
result = await coro
print(result)
asyncio.run(main())
# 输出顺序: task2, task3, task1四、实际应用：并发爬虫
import asyncio
import aiohttp
async def fetch_page(session, url):
async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
text = await resp.text()
return {
"url": url,
"status": resp.status,
"length": len(text),
}
async def crawl(urls, concurrency=5):
semaphore = asyncio.Semaphore(concurrency)
async def limited_fetch(session, url):
async with semaphore:
return await fetch_page(session, url)
async with aiohttp.ClientSession() as session:
tasks = [limited_fetch(session, url) for url in urls]
return await asyncio.gather(*tasks, return_exceptions=True)
urls = [f"https://httpbin.org/delay/{i % 3}" for i in range(20)]
results = asyncio.run(crawl(urls, concurrency=5))
for r in results:
if isinstance(r, Exception):
print(f"Error: {r}")
else:
print(f"{r[\"status\"]} - {r[\"url\"]} ({r[\"length\"]} bytes)")五、异步上下文管理器
class AsyncDBConnection:
async def __aenter__(self):
print("连接数据库...")
await asyncio.sleep(0.5)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
print("关闭连接...")
await asyncio.sleep(0.1)
async def query(self, sql):
await asyncio.sleep(0.1)
return f"Result of: {sql}"
async def main():
async with AsyncDBConnection() as db:
result = await db.query("SELECT * FROM users")
print(result)
asyncio.run(main())六、常见陷阱
6.1 在异步中调用同步阻塞代码
# 错误：requests 是同步库，会阻塞事件循环
async def bad_fetch(url):
return requests.get(url).text
# 正确：使用 aiohttp
async def good_fetch(url):
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
return await resp.text()6.2 忘记 await
async def main():
asyncio.sleep(1) # 不会执行！返回 coroutine 对象
await asyncio.sleep(1) # 正确6.3 在异步代码中使用 time.sleep
import time
async def main():
time.sleep(1) # 阻塞事件循环！
await asyncio.sleep(1) # 正确：让出控制权七、总结
asyncio 的核心价值在于：在单线程内通过事件循环实现高并发 IO，避免了多线程的锁问题和上下文切换开销。记住三个关键点：IO 操作必须用异步库（aiohttp 而非 requests）、用 gather/create_task 实现并发、用 Semaphore 控制并发量。在高 IO 场景（爬虫、API 调用、数据库查询）中，asyncio 能带来数量级的性能提升。
评论 (0)