Python 异步编程与协程实战指南
在 I/O 密集型任务中，同步阻塞往往成为性能瓶颈。异步编程允许程序在等待网络或文件操作时继续处理其他逻辑，从而显著提升吞吐量。本文将深入探讨 Python 中的协程、事件循环以及 asyncio 和 aiohttp 库的实战应用。
为什么需要异步编程
传统的同步代码在执行耗时操作（如 HTTP 请求）时会挂起当前线程，导致资源浪费。异步模式通过非阻塞机制，让单个线程能并发处理多个任务。其核心优势在于：
- 提升效率：I/O 等待期间不占用 CPU 时间片。
- 降低开销：相比多线程，协程切换成本极低。
- 简化结构：避免回调地狱，代码更接近同步写法。
常见场景包括高并发 Web 服务、批量数据抓取及实时通信等。
理解协程：轻量级并发单元
协程（Coroutine）是 Python 异步编程的基础。使用 async def 定义的函数即为协程函数，调用它不会立即执行，而是返回一个协程对象。真正的执行需要通过事件循环调度。
import asyncio
async def hello():
print('Hello, World!')
await asyncio.sleep(1)
print('Hello again!')
注意 await 关键字，它表示在此处暂停协程，将控制权交还给事件循环，直到 sleep 完成。若需同时运行多个协程，可使用 asyncio.gather：
async def count():
print('Counting...')
await asyncio.sleep(1)
print('Counted!')
async def main():
await asyncio.gather(count(), count(), count())
asyncio.run(main())
任务调度与管理
在实际开发中，我们常需要将协程包装为任务（Task），以便更灵活地控制生命周期。
创建与取消任务
使用 asyncio.create_task 可以将协程注册到事件循环中，使其后台运行。若需提前终止，可调用 task.cancel()。
import asyncio
async def hello():
try:
print('Hello, World!')
await asyncio.sleep(1)
print('Hello again!')
except asyncio.CancelledError:
print('Task cancelled!')
async def main():
task = asyncio.create_task(hello())
await asyncio.sleep(0.5)
task.cancel()
try:
await task
except asyncio.CancelledError:
print('Main: Task cancelled!')
asyncio.run(main())
设置超时保护
防止某个任务无限期挂起，可以使用 asyncio.wait_for 设置截止时间。
async def hello():
print('Hello, World!')
await asyncio.sleep(2)
print('Hello again!')
async def main():
try:
await asyncio.wait_for(hello(), timeout=1)
except asyncio.TimeoutError:
print('Task timed out!')
asyncio.run(main())
事件循环与 asyncio 核心
事件循环是异步程序的引擎，负责轮询并调度待执行的任务。虽然现代 Python 推荐使用 asyncio.run() 自动管理循环，但了解底层机制有助于调试。
import asyncio
loop = asyncio.get_event_loop()
async def hello():
print('Hello, World!')
await asyncio.sleep(1)
print('Hello again!')
loop.run_until_complete(hello())
常用工具函数包括：
asyncio.run()：启动新事件循环并运行主协程。asyncio.create_task()：创建后台任务。asyncio.gather()：并行收集多个任务结果。asyncio.sleep()：模拟异步等待。
对于更复杂的场景，还可以利用 Future 对象手动控制状态流转。
基于 aiohttp 的网络编程
aiohttp 是 Python 生态中最成熟的异步 HTTP 客户端与服务端库。
安装依赖
pip install aiohttp
构建异步 HTTP 客户端
发送请求时，无需阻塞主线程。下面示例展示了如何并发抓取多个 URL 并统计耗时。
import aiohttp
import asyncio
import time
async def fetch(session, url):
start_time = time.time()
async with session.get(url) as response:
text = await response.text()
elapsed_time = time.time() - start_time
return url, len(text), elapsed_time
async def main():
urls = [
'https://www.example.com',
'https://www.google.com',
'https://www.github.com'
]
async with aiohttp.ClientSession() as session:
tasks = [asyncio.create_task(fetch(session, url)) for url in urls]
results = await asyncio.gather(*tasks)
for url, length, elapsed_time in results:
print(f'URL: {url}, 响应长度：{length}, 耗时：{elapsed_time:.2f}秒')
if __name__ == '__main__':
start_time = time.time()
asyncio.run(main())
elapsed_time = time.time() - start_time
print(f'总耗时：{elapsed_time:.2f}秒')
搭建异步 Web 服务器
除了作为客户端，aiohttp 也能快速构建高性能服务端。
from aiohttp import web
import asyncio
async def handle_root(request):
return web.Response(text='Hello, World!')
async def handle_api(request):
data = {'name': '张三', 'age': 25}
return web.json_response(data)
async def create_app():
app = web.Application()
app.add_routes([
web.get('/', handle_root),
web.get('/api', handle_api),
])
return app
if __name__ == '__main__':
asyncio.run(web.run_app(create_app(), host='localhost', port=8080))
总结
掌握异步编程的关键在于理解事件循环的工作机制，并熟练运用 asyncio 提供的工具。通过 aiohttp，我们可以轻松实现高并发的网络交互。建议在实际项目中多尝试编写异步代码，体会其对资源调度的优化效果。