Python 异步编程与协程实战指南
异步编程概述
异步编程是一种通过非阻塞操作提升程序执行效率的并发方式。在等待 I/O 操作（如网络请求、文件读写）完成时，程序不会挂起，而是继续处理其他任务。这种机制特别适用于高并发的网络通信、大文件处理及数据库查询场景。
相比传统线程模型，异步编程能显著降低资源消耗，减少线程切换开销，并通过协程和任务调度让代码结构更加简洁。
协程的定义与使用
协程（Coroutine）是轻量级的用户态线程，支持暂停和恢复执行。在 Python 中，我们使用 async def 关键字来定义协程函数。
基本协程示例
import asyncio
async def hello():
print('Hello, World!')
await asyncio.sleep(1)
print('Hello again!')
# 运行协程
asyncio.run(hello())
注意 await 关键字，它表示当前协程在此处暂停，直到被等待的对象完成。多个协程可以通过 asyncio.gather() 并发执行：
import asyncio
async def count():
print('Counting...')
await asyncio.sleep(1)
print('Counted!')
async def main():
await asyncio.gather(count(), count(), count())
asyncio.run(main())
任务调度管理
在实际开发中，我们需要对协程进行更精细的控制，包括创建任务、取消任务以及设置超时。
创建与管理任务
使用 asyncio.create_task() 可以将协程封装为 Task 对象，便于后续管理和取消。
import asyncio
async def hello():
print('Hello, World!')
await asyncio.sleep(1)
print('Hello again!')
async def main():
task1 = asyncio.create_task(hello())
task2 = asyncio.create_task(hello())
await task1
await task2
asyncio.run(main())
任务取消与超时
处理异常是健壮性的关键。如果任务需要被取消，或者执行时间过长，我们需要捕获相应的异常。
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() 非常实用，它能确保任务在规定时间内完成，否则抛出 TimeoutError。
import asyncio
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())
事件循环核心
事件循环（Event Loop）是异步编程的心脏，负责从任务队列中取出任务并调度执行。虽然现代 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 库详解
asyncio 是 Python 标准库中实现异步 I/O 的核心模块。
常用函数速查
asyncio.run(): 启动事件循环，运行顶层协程。asyncio.create_task(): 创建后台任务。asyncio.gather(): 收集多个协程的结果。asyncio.wait_for(): 带超时的等待。asyncio.sleep(): 模拟异步延迟。
高级用法：Future 对象
在处理回调逻辑时，Future 对象提供了灵活的状态通知机制。
import asyncio
async def hello():
print('Hello, World!')
await asyncio.sleep(1)
print('Hello again!')
async def main():
future = asyncio.Future()
task = asyncio.create_task(hello())
task.add_done_callback(lambda t: future.set_result(t.result()))
await future
asyncio.run(main())
aiohttp 库实战
aiohttp 是基于 asyncio 的 HTTP 客户端和服务端库，非常适合构建高性能 Web 应用。
安装与基础请求
首先安装依赖：
pip install aiohttp
发送 GET 请求示例：
import aiohttp
import asyncio
async def fetch(session, url):
async with session.get(url) as response:
return await response.text()
async def main():
async with aiohttp.ClientSession() as session:
html = await fetch(session, 'https://www.example.com')
print(html)
asyncio.run(main())
发送 POST 请求和 JSON 数据同样简单：
import aiohttp
import asyncio
import json
async def post_json(session, url, data):
async with session.post(url, json=data) as response:
return await response.text()
async def main():
async with aiohttp.ClientSession() as session:
data = {'name': '张三', 'age': 25}
response = await post_json(session, 'https://httpbin.org/post', data)
print(response)
asyncio.run(main())
实战案例：高并发 HTTP 客户端
下面是一个完整的异步 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',
'https://www.python.org',
'https://www.djangoproject.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.web 模块，我们可以快速搭建一个异步 Web 服务。
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 handle_static(request):
return web.FileResponse('static/index.html')
async def create_app():
app = web.Application()
app.add_routes([
web.get('/', handle_root),
web.get('/api', handle_api),
web.get('/static/{name}', handle_static)
])
return app
if __name__ == '__main__':
asyncio.run(web.run_app(create_app(), host='localhost', port=8080))
该服务器能够处理根路径访问、返回 JSON API 数据以及提供静态文件服务，所有请求处理均是非阻塞的。
总结
掌握 Python 异步编程的关键在于理解事件循环与协程的配合。通过 asyncio 和 aiohttp，开发者可以编写出高效、低资源消耗的并发程序。无论是构建高并发爬虫还是微服务接口，异步模式都是现代 Python 开发不可或缺的技能。建议在实际项目中多尝试结合具体业务场景练习，深入体会其性能优势。