一、异步编程的核心原理
1.1 什么是异步编程？
传统的同步编程中，代码按照顺序一行行执行，遇到 IO 操作（如网络请求、文件读写）时，程序会阻塞等待操作完成，导致 CPU 空闲浪费。而异步编程的核心思想是：遇到 IO 操作时自动切换，IO 操作完成后自动切回，在单线程内实现高并发。
1.2 协程的工作原理
协程是异步编程的基础单元，它的工作流程如下：
┌─────────────────────────────────────────────────────────┐
│ 事件循环 (Event Loop)
├─────────────────────────────────────────────────────────┤
│
│ ┌──────────┐ ┌──────────┐ ┌──────────┐
│ │ 任务 1 │ ──> │ 任务 2 │ ──> │ 任务 3 │
│ │ (协程 A) │ │ (协程 B) │ │ (协程 C) │
│ └────┬─────┘ └────┬─────┘ └────┬─────┘
│ │ │ │
│ ▼ await ▼ await ▼ await
│ ┌──────────┐ ┌──────────┐ ┌──────────┐
│ │ IO 操作 1 │ │ IO 操作 2 │ │ IO 操作 3 │
│ └──────────┘ └──────────┘ └──────────┘
│ │ │ │
│ └────────────────┼────────────────┘
│ ▼
│ IO 完成，回调通知
│
└─────────────────────────────────────────────────────────┘
关键机制：
- 遇到 IO 自动挂起：执行到
await关键字时，协程主动让出 CPU - 事件循环调度：事件循环负责管理所有协程，当某个协程挂起时，立即切换到下一个就绪协程
- IO 完成自动恢复：底层通过
select/epoll等机制监听 IO 事件，完成后将对应协程放回就绪队列
二、Python 异步编程的演进
2.1 发展阶段对比
| 版本 | 核心特性 | 代表作 | 缺点 |
|---|---|---|---|
| Python 2.x | 无原生支持 | gevent（第三方） | 猴子补丁，魔法太重 |
| Python 3.3 | yield from | asyncio 雏形 | 生成器协程混淆 |
| Python 3.5 | async/await | 原生协程 | - |
| Python 3.7+ | asyncio.run() | 极致简化 | - |
2.2 原生协程的优势
Python 3.5 引入的 async/await 语法彻底改变了异步编程：
# 生成器协程（3.4 及以前）
@asyncio.coroutine
def hello():
yield from asyncio.sleep(1)
print('Hello')
# 原生协程（3.5+）
async def hello():
await asyncio.sleep(1)
print('Hello')
# 更清晰，无混淆
三、asyncio 核心组件详解
3.1 四大核心对象
import asyncio
# 1. 协程函数和协程对象
async def coro_func():
return 42
coro = coro_func() # 协程对象（此时未执行）
# 2. 任务（Task）- 对协程的封装
task = asyncio.create_task(coro()) # Python 3.7+ 推荐方式
# 或 task = asyncio.ensure_future(coro())
# 3. 事件循环 - 调度器
loop = asyncio.get_event_loop() # 获取事件循环
loop.run_until_complete(coro()) # 运行直到完成
# Python 3.7+ 简单方式：
asyncio.run(coro()) # 自动创建/关闭事件循环
3.2 awaitable 对象的三种类型
# 1. 协程（Coroutine）
async def foo():
return 123
# 2. 任务（Task）
async def main():
task = asyncio.create_task(foo())
result = await task
# 3. Future（底层对象，通常不直接使用）
fut = asyncio.Future()
asyncio.ensure_future(set_after(fut, 1, 456))
result = await fut
四、实战：高并发 HTTP 请求
4.1 基础用法
import asyncio
import aiohttp
import time
async def fetch_one(session, url):
"""单个 HTTP 请求"""
async with session.get(url) as response:
# await 挂起直到响应返回
return await response.text()
async def main_simple():
"""简单示例：请求单个 URL"""
async with aiohttp.ClientSession() as session:
html = await fetch_one(session, 'http://httpbin.org/get')
print(f"响应长度：{len(html)}")
# Python 3.7+ 运行方式
asyncio.run(main_simple())
4.2 高并发批量请求
这是异步编程发挥威力的核心场景：
import asyncio
import aiohttp
import time
from typing import List, Dict
async def fetch_url(session: aiohttp.ClientSession, url: str) -> Dict:
"""单个 URL 请求，带错误处理"""
start = time.time()
try:
async with session.get(url, timeout=10) as response:
content = await response.text()
return {
'url': url,
'status': response.status,
'length': len(content),
'time': time.time() - start,
'success': True
}
except Exception as e:
return {
'url': url,
'error': str(e),
'time': time.time() - start,
'success': False
}
async def fetch_many(urls: List[str], max_concurrent: int = 10):
"""
高并发请求多个 URL
- 使用信号量控制并发数
- 收集所有结果
"""
# 控制并发量的信号量
semaphore = asyncio.Semaphore(max_concurrent)
async def bounded_fetch(url):
async with semaphore:
# 限制并发数
return await fetch_url(session, url)
async with aiohttp.ClientSession() as session:
# 创建所有任务
tasks = [bounded_fetch(url) for url in urls]
# 并发执行并收集结果
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
async def main():
"""性能对比演示"""
# 测试 URL 列表（10 个不同请求）
urls = [
'http://httpbin.org/delay/1', # 延迟 1 秒
'http://httpbin.org/get',
'http://httpbin.org/json',
'http://httpbin.org/xml',
'http://httpbin.org/robots.txt',
'http://httpbin.org/anything',
'http://httpbin.org/uuid',
'http://httpbin.org/image',
'http://httpbin.org/headers',
'http://httpbin.org/ip'
] * 2 # 20 个请求
# 1. 异步并发执行
start = time.time()
results = await fetch_many(urls, max_concurrent=5)
async_time = time.time() - start
# 2. 统计结果
success_count = sum(1 for r in results if isinstance(r, dict) and r.get('success'))
print(f"异步并发请求 {len(urls)} 个 URL:")
print(f" 耗时：{async_time:.2f}秒")
print(f" 成功：{success_count}/{len(results)}")
# 3. 如果要对比同步版本（耗时通常是异步的 5-10 倍）
# 请使用 requests 库顺序执行做对比测试
if __name__ == "__main__":
asyncio.run(main())
4.3 高级模式：生产者 - 消费者
import asyncio
import aiohttp
from asyncio import Queue
async def producer(queue: Queue, urls: List[str]):
"""生产者：将 URL 放入队列"""
for url in urls:
await queue.put(url)
print(f"生产者放入：{url}")
# 发送结束信号
for _ in range(3): # 消费者数量
await queue.put(None)
async def consumer(queue: Queue, session: aiohttp.ClientSession, name: str):
"""消费者：从队列取 URL 并请求"""
while True:
url = await queue.get()
if url is None:
queue.task_done()
break
print(f"消费者{name} 处理：{url}")
try:
async with session.get(url) as resp:
text = await resp.text()
print(f"消费者{name} 完成：{url}, 大小：{len(text)}")
except Exception as e:
print(f"消费者{name} 失败：{url}, 错误：{e}")
finally:
queue.task_done()
async def producer_consumer_demo():
"""生产者 - 消费者模式示例"""
urls = ['http://httpbin.org/get'] * 20
queue = Queue(maxsize=5) # 缓冲队列
async with aiohttp.ClientSession() as session:
# 启动消费者
consumers = [
asyncio.create_task(consumer(queue, session, f"{i}"))
for i in range(3)
]
# 启动生产者
producer_task = asyncio.create_task(producer(queue, urls))
# 等待所有任务完成
await asyncio.gather(producer_task, *consumers)
await queue.join() # 等待队列清空
asyncio.run(producer_consumer_demo())
五、异步上下文管理器与异步迭代器
5.1 异步上下文管理器（async with）
class AsyncResource:
"""模拟需要异步初始化和清理的资源"""
async def __aenter__(self):
print("正在获取资源...")
await asyncio.sleep(1) # 模拟异步初始化
print("资源已获取")
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
print("正在释放资源...")
await asyncio.sleep(0.5) # 模拟异步清理
print("资源已释放")
async def work(self):
return "资源使用中"
async def use_async_context():
async with AsyncResource() as resource:
result = await resource.work()
print(result)
# 实际应用：数据库连接
class DatabaseConnection:
async def __aenter__(self):
self.conn = await create_db_connection()
return self.conn
async def __aexit__(self, *args):
await self.conn.close()
async def query_db():
async with DatabaseConnection() as conn:
return await conn.execute("SELECT * FROM users")
5.2 异步迭代器（async for）
import asyncio
class AsyncRange:
"""异步范围迭代器"""
def __init__(self, start, end, delay=0.1):
self.start = start
self.end = end
self.delay = delay
self.current = start
def __aiter__(self):
return self
async def __anext__(self):
if self.current >= self.end:
raise StopAsyncIteration
await asyncio.sleep(self.delay) # 模拟异步操作
self.current += 1
return self.current - 1
async def main():
async for num in AsyncRange(1, 5):
print(f"异步生成：{num}")
asyncio.run(main())
# 实际应用：分页 API 请求
class PaginatedAPI:
async def __aiter__(self):
return self
async def __anext__(self):
page = await self.fetch_page()
if not page:
raise StopAsyncIteration
return page
async def fetch_page(self):
# 实际 API 请求逻辑
pass
async def fetch_all_pages():
async for page in PaginatedAPI():
await process_page(page)
六、性能优化与最佳实践
6.1 七项核心原则
# 1. 正确创建任务
async def good():
task = asyncio.create_task(coro()) # 立即调度
await task
# 错误：协程不会并发执行
async def bad():
await coro() # 等价于同步调用
await coro2()
# 2. 使用 gather/wait 并发收集
async def fetch_all():
results = await asyncio.gather(
fetch_url(url1),
fetch_url(url2),
return_exceptions=True # 防止单个失败影响全部
)
# 3. 控制并发数量（信号量）
sem = asyncio.Semaphore(10)
async def bounded_fetch(url):
async with sem:
return await fetch_url(url)
# 4. 设置超时
async def fetch_with_timeout(url):
try:
return await asyncio.wait_for(fetch_url(url), timeout=5.0)
except asyncio.TimeoutError:
return None
# 5. 使用异步库，不用阻塞调用
# aiohttp | requests
# aiomysql | pymysql
# asyncpg | psycopg2
# 6. 避免跨线程使用事件循环
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
# 然后在子线程执行 loop.run_forever()
# 7. 正确处理取消
async def cancellable():
try:
await asyncio.sleep(10)
except asyncio.CancelledError:
print("任务被取消")
await cleanup()
raise # 重新抛出
6.2 性能对比测试框架
import asyncio
import aiohttp
import requests
import time
from functools import wraps
def timeit(func):
"""性能计时装饰器"""
@wraps(func)
async def async_wrapper(*args, **kwargs):
start = time.perf_counter()
result = await func(*args, **kwargs)
cost = time.perf_counter() - start
print(f"{func.__name__} 耗时：{cost:.3f}秒")
return result
return async_wrapper
@timeit
async def async_benchmark():
"""异步版本性能测试"""
urls = ['http://httpbin.org/get'] * 20
async with aiohttp.ClientSession() as session:
tasks = [fetch_url(session, url) for url in urls]
return await asyncio.gather(*tasks)
def sync_benchmark():
"""同步版本性能测试（对比用）"""
urls = ['http://httpbin.org/get'] * 20
results = []
start = time.perf_counter()
for url in urls:
results.append(requests.get(url).text)
print(f"sync_benchmark 耗时：{time.perf_counter() - start:.3f}秒")
return results
# 运行对比测试
asyncio.run(async_benchmark())
sync_benchmark()
七、常见陷阱与解决方案
7.1 常见错误排查
# 1. 忘记 await
async def mistake1():
coro() # 协程未执行，警告：coroutine never awaited
# 正确
async def correct1():
await coro()
# 2. 在同步函数中创建事件循环多次
def mistake2():
asyncio.run(main()) # OK
asyncio.run(main()) # 事件循环已关闭
# 正确：一个程序只有一个入口
if __name__ == "__main__":
asyncio.run(main())
# 3. 阻塞事件循环
async def mistake3():
time.sleep(1) # 阻塞所有协程！
await asyncio.sleep(0)
# 正确
async def correct3():
await asyncio.sleep(1)
# 4. 任务创建后不 await
async def mistake4():
asyncio.create_task(work()) # 任务可能未完成就结束
# 程序退出，任务被取消
# 正确
async def correct4():
task = asyncio.create_task(work())
await task # 或 await asyncio.gather(task)
7.2 调试技巧
import asyncio
import logging
# 启用调试日志
logging.basicConfig(level=logging.DEBUG)
# 开启 asyncio 调试模式
asyncio.run(main(), debug=True)
# 或手动设置
loop = asyncio.get_event_loop()
loop.set_debug(True)
# 查看未等待的任务
pending = asyncio.all_tasks(loop)
for task in pending:
print(f"未完成任务：{task}")
八、总结与展望
8.1 异步编程的核心要点
- 原理理解：协程遇 IO 自动切换，事件循环统一调度
- 语法掌握：
async/await、async with、async for、asyncio.gather() - 库选择：使用 aiohttp、asyncpg 等原生异步库
- 模式应用：信号量限流、生产者 - 消费者、超时控制
- 性能意识：避免阻塞调用，合理设置并发数
8.2 适用场景
| 场景 | 推荐度 | 原因 |
|---|---|---|
| 网络爬虫 | 五星 | 大量 IO 等待，异步收益明显 |
| Web 应用 | 五星 | FastAPI、Sanic 等框架原生支持 |
| 数据库访问 | 四星 | 连接池 + 异步驱动，吞吐量提升 |
| CPU 密集型 | 两星 | 多进程更合适 |
| 简单脚本 | 三星 | 视 IO 密集程度而定 |
| · |
8.3 未来演进
Python 3.11+ 引入了更高效的 asyncio 实现，性能进一步提升。异步编程已成为 Python 生态中处理高并发 IO 任务的标准方案，掌握它是现代 Python 开发者必备的核心技能。