Python 异步编程与协程实战指南

任务调度管理

在实际开发中，我们往往需要更精细地控制任务的创建、取消和超时。

创建与取消任务

使用 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())

注意捕获 CancelledError，否则任务被取消时可能会抛出未处理的异常。

设置超时

对于可能卡死的任务，务必加上超时保护：

import asyncio

async def hello():
    print('Hello, World!')
    await asyncio.sleep(2)
    print('Hello again!')

async def main():
    try:
        # 限制执行时间为 1 秒
        await asyncio.wait_for(hello(), timeout=1)
    except asyncio.TimeoutError:
        print('Task timed out!')

asyncio.run(main())

事件循环机制

事件循环是异步编程的心脏，负责不断从队列中取出任务并执行。虽然现代 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 提供了丰富的工具函数，以下是常用列表：

• asyncio.run()：启动事件循环，运行顶层协程。
• asyncio.create_task()：创建后台任务。
• asyncio.gather()：并发执行多个协程并收集结果。
• asyncio.wait_for()：带超时的等待。
• asyncio.sleep()：模拟异步休眠。

高级用法中，还可以结合 Future 对象进行回调处理：

import asyncio

async def hello():
    return 'Done'

async def main():
    future = asyncio.Future()
    task = asyncio.create_task(hello())
    task.add_done_callback(lambda t: future.set_result(t.result()))
    result = await future
    print(result)

asyncio.run(main())

aiohttp 库实战

处理网络请求时，aiohttp 是最常用的异步 HTTP 客户端和服务端库。

安装依赖

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}秒')

这段代码展示了如何利用 gather 批量处理请求，相比串行执行，总耗时将接近最慢的那个请求时间，而非所有请求之和。

实战案例：异步 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))

这里定义了根路径、API 接口和静态文件服务，路由配置清晰，适合快速搭建微服务接口。

总结

掌握 Python 异步编程的关键在于理解事件循环和协程的生命周期。通过 asyncio 和 aiohttp，我们可以轻松构建高效的网络应用。建议在实际项目中多尝试并发场景，比如批量爬虫或 API 聚合服务，这样能更深刻地体会到异步带来的性能提升。