Python 异步编程与 asyncio 完全指南
折
折腾侠
2026/04/26 发布
0约 8 分钟1330 字 / 846 词00
Python 异步编程与 asyncio 完全指南
引言
在现代软件开发中，性能和高并发处理能力是衡量应用程序质量的重要指标。Python 作为一门广泛使用的编程语言，其异步编程能力通过 INLINE_CODE_0 库得到了极大的提升。本教程将深入讲解 Python 异步编程的核心概念、工作原理以及实际应用场景，帮助你掌握这一重要技能。
什么是异步编程？
异步编程是一种编程范式，允许程序在等待某些操作（如 I/O 操作、网络请求、数据库查询）完成时，不阻塞主线程的执行，而是转而去执行其他任务。当等待的操作完成后，程序会自动恢复执行。
与传统的同步编程相比，异步编程的主要优势在于：
- 更高的并发性能：单个线程可以处理多个并发任务
- 更好的资源利用率：减少线程切换开销
- 更简洁的代码结构：避免回调地狱（Callback Hell）
asyncio 核心概念
1. async 和 await
INLINE_CODE_1 关键字用于定义协程函数，INLINE_CODE_2 用于等待异步操作完成。
Python
import asyncio
async def fetch_data(url):
"""模拟异步数据获取"""
print(f"开始获取数据：{url}")
await asyncio.sleep(2) # 模拟网络延迟
print(f"数据获取完成：{url}")
return {"url": url, "status": "success"}
async def main():
result = await fetch_data("https://api.example.com/data")
print(result)
asyncio.run(main())
2. Event Loop（事件循环）
事件循环是 asyncio 的核心，它负责调度和执行协程。事件循环不断检查是否有就绪的任务，并执行它们。
Python
import asyncio
async def task(name, delay):
for i in range(3):
print(f"{name} - 执行第 {i+1} 次")
await asyncio.sleep(delay)
print(f"{name} - 完成")
async def main():
# 创建事件循环并运行多个任务
await asyncio.gather(
task("任务 A", 1),
task("任务 B", 1.5),
task("任务 C", 0.8)
)
asyncio.run(main())
3. Task（任务）
Task 是 asyncio 中用于并发执行协程的包装器。通过创建 Task，可以让多个协程同时运行。
Python
import asyncio
async def worker(id, delay):
print(f"Worker {id} 开始工作")
await asyncio.sleep(delay)
print(f"Worker {id} 完成工作")
return f"Worker {id} 的结果"
async def main():
# 创建多个任务并发执行
tasks = [
asyncio.create_task(worker(i, 2))
for i in range(5)
]
# 等待所有任务完成
results = await asyncio.gather(*tasks)
print("所有任务完成:", results)
asyncio.run(main())
实际应用场景
场景一：并发网络请求
在爬虫或 API 聚合服务中，经常需要同时发起多个网络请求。使用 asyncio 可以显著提高性能。
Python
import asyncio
import aiohttp
async def fetch_url(session, url):
"""异步获取单个 URL 内容"""
try:
async with session.get(url, timeout=10) as response:
content = await response.text()
return {
"url": url,
"status": response.status,
"length": len(content)
}
except Exception as e:
return {"url": url, "error": str(e)}
async def fetch_all_urls(urls, max_concurrent=10):
"""并发获取多个 URL"""
async with aiohttp.ClientSession() as session:
# 使用信号量控制并发数量
semaphore = asyncio.Semaphore(max_concurrent)
async def bounded_fetch(url):
async with semaphore:
return await fetch_url(session, url)
tasks = [bounded_fetch(url) for url in urls]
results = await asyncio.gather(*tasks)
return results
async def main():
urls = [
"https://www.python.org",
"https://www.github.com",
"https://www.stackoverflow.com",
"https://www.reddit.com",
"https://www.wikipedia.org"
]
results = await fetch_all_urls(urls)
for result in results:
if "error" in result:
print(f"错误：{result['url']} - {result['error']}")
else:
print(f"成功：{result['url']} - 状态 {result['status']}, 长度 {result['length']}")
# 运行需要安装 aiohttp: pip install aiohttp
# asyncio.run(main())
场景二：异步数据库操作
使用异步数据库驱动可以显著提高数据库密集型应用的性能。
Python
import asyncio
import asyncpg
async def query_database():
"""异步数据库查询示例"""
# 连接数据库
conn = await asyncpg.connect(
host='localhost',
port=5432,
user='postgres',
password='password',
database='mydb'
)
try:
# 并发执行多个查询
results = await asyncio.gather(
conn.fetch("SELECT * FROM users WHERE active = $1", True),
conn.fetch("SELECT * FROM orders WHERE status = $1", 'pending'),
conn.fetch("SELECT * FROM products WHERE stock > $1", 0)
)
users, orders, products = results
print(f"活跃用户：{len(users)}")
print(f"待处理订单：{len(orders)}")
print(f"有库存商品：{len(products)}")
finally:
await conn.close()
# 运行需要安装 asyncpg: pip install asyncpg
# asyncio.run(query_database())
场景三：异步文件处理
处理大量文件时，异步 I/O 可以避免阻塞主线程。
Python
import asyncio
import aiofiles
async def process_file(filepath):
"""异步处理单个文件"""
async with aiofiles.open(filepath, 'r') as f:
content = await f.read()
# 模拟处理逻辑
await asyncio.sleep(0.5)
word_count = len(content.split())
return {"file": filepath, "words": word_count}
async def process_multiple_files(filepaths):
"""并发处理多个文件"""
tasks = [process_file(fp) for fp in filepaths]
results = await asyncio.gather(*tasks)
total_words = sum(r["words"] for r in results)
print(f"处理了 {len(results)} 个文件，总计 {total_words} 个单词")
return results
# 运行需要安装 aiofiles: pip install aiofiles
# file_list = ["file1.txt", "file2.txt", "file3.txt"]
# asyncio.run(process_multiple_files(file_list))
场景四：WebSocket 实时通信
异步编程非常适合处理 WebSocket 这类长连接场景。
Python
import asyncio
import websockets
async def websocket_handler(websocket, path):
"""处理 WebSocket 连接"""
print(f"新连接：{websocket.remote_address}")
try:
async for message in websocket:
print(f"收到消息：{message}")
# 处理消息并回复
response = f"服务器收到：{message}"
await websocket.send(response)
except websockets.exceptions.ConnectionClosed:
print("连接关闭")
async def main():
# 启动 WebSocket 服务器
server = await websockets.serve(
websocket_handler,
"localhost",
8765
)
print("WebSocket 服务器启动在 ws://localhost:8765")
await server.wait_closed()
# 运行需要安装 websockets: pip install websockets
# asyncio.run(main())
高级技巧与最佳实践
1. 超时控制
使用 INLINE_CODE_3 为异步操作设置超时。
Python
import asyncio
async def slow_operation():
await asyncio.sleep(10)
return "完成"
async def main():
try:
# 设置 5 秒超时
result = await asyncio.wait_for(
slow_operation(),
timeout=5.0
)
print(result)
except asyncio.TimeoutError:
print("操作超时！")
asyncio.run(main())
2. 错误处理
异步代码中的错误处理需要特别注意。
Python
import asyncio
async def risky_task(id):
if id == 2:
raise ValueError(f"任务 {id} 出错了")
await asyncio.sleep(1)
return f"任务 {id} 成功"
async def main():
tasks = [asyncio.create_task(risky_task(i)) for i in range(5)]
# 使用 return_exceptions 捕获异常而不中断其他任务
results = await asyncio.gather(*tasks, return_exceptions=True)
for i, result in enumerate(results):
if isinstance(result, Exception):
print(f"任务 {i} 失败：{result}")
else:
print(f"任务 {i} 成功：{result}")
asyncio.run(main())
3. 生产者 - 消费者模式
使用 Queue 实现异步生产者 - 消费者模式。
Python
import asyncio
async def producer(queue, id):
for i in range(5):
item = f"生产者 {id} - 物品 {i}"
await queue.put(item)
print(f"生产：{item}")
await asyncio.sleep(0.5)
await queue.put(None) # 结束标记
async def consumer(queue, id):
while True:
item = await queue.get()
if item is None:
queue.task_done()
break
await asyncio.sleep(0.3)
print(f"消费者 {id} 处理：{item}")
queue.task_done()
async def main():
queue = asyncio.Queue(maxsize=10)
# 创建生产者和消费者
producers = [asyncio.create_task(producer(queue, i)) for i in range(2)]
consumers = [asyncio.create_task(consumer(queue, i)) for i in range(3)]
# 等待生产者完成
await asyncio.gather(*producers)
# 等待队列处理完成
await queue.join()
# 等待消费者完成
await asyncio.gather(*consumers)
print("所有任务完成！")
asyncio.run(main())
性能对比
让我们看看异步编程相比传统多线程的性能优势：
Python
import asyncio
import threading
import time
# 同步版本
def sync_fetch(url_id):
time.sleep(1) # 模拟 I/O 延迟
return f"URL {url_id}"
def sync_main():
start = time.time()
results = [sync_fetch(i) for i in range(10)]
end = time.time()
print(f"同步执行时间：{end - start:.2f} 秒")
# 异步版本
async def async_fetch(url_id):
await asyncio.sleep(1)
return f"URL {url_id}"
async def async_main():
start = time.time()
tasks = [async_fetch(i) for i in range(10)]
results = await asyncio.gather(*tasks)
end = time.time()
print(f"异步执行时间：{end - start:.2f} 秒")
# 运行对比
# sync_main() # 约 10 秒
# asyncio.run(async_main()) # 约 1 秒
从上面的对比可以看出，对于 I/O 密集型任务，异步编程可以将执行时间从 10 秒降低到 1 秒，性能提升 10 倍！
常见陷阱与注意事项
-
避免阻塞调用：在 async 函数中不要使用同步的 I/O 操作（如 INLINE_CODE_4、INLINE_CODE_5），这会阻塞事件循环。
-
正确使用 asyncio.gather：INLINE_CODE_6 默认会在第一个任务失败时抛出异常，使用 INLINE_CODE_7 可以收集所有结果。
-
注意内存泄漏：长时间运行的任务要确保正确清理资源，使用 INLINE_CODE_8 或异步上下文管理器。
-
调试技巧：使用 INLINE_CODE_9 时，保留 task 引用以防任务被垃圾回收。
结语
Python 的异步编程能力通过 asyncio 库变得强大而易用。掌握异步编程可以显著提升你的应用程序性能，特别是在处理大量 I/O 操作、网络请求或并发任务时。
本教程涵盖了异步编程的核心概念、常用模式和实际应用场景。建议你从简单的例子开始实践，逐步掌握这一重要技能。记住，异步编程的关键在于理解事件循环的工作原理，并学会正确地使用 async/await 语法。
随着 Python 生态系统的不断发展，越来越多的库都提供了异步支持（如 aiohttp、asyncpg、motor 等），这使得构建高性能的异步应用变得更加容易。开始你的异步编程之旅吧！