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

任务调度与管理

创建与取消任务

使用 asyncio.create_task 将协程包装为 Task 对象，便于管理生命周期。若需提前终止任务，可调用 cancel() 方法，并捕获 CancelledError 进行清理。

import asyncio

async def task_work():
    try:
        print('Task started')
        await asyncio.sleep(10)
        print('Task finished')
    except asyncio.CancelledError:
        print('Task cancelled gracefully')

async def main():
    t = asyncio.create_task(task_work())
    await asyncio.sleep(0.5)
    t.cancel()
    try:
        await t
    except asyncio.CancelledError:
        print('Main: Task cancelled!')

asyncio.run(main())

超时控制

对于可能卡死的任务，建议设置超时限制。asyncio.wait_for 能在指定时间后抛出 TimeoutError。

import asyncio

async def slow_op():
    await asyncio.sleep(5)

async def main():
    try:
        await asyncio.wait_for(slow_op(), timeout=1)
    except asyncio.TimeoutError:
        print('Operation timed out')

asyncio.run(main())

事件循环原理

事件循环（Event Loop）是异步引擎的核心，负责轮询就绪的任务并分发执行。虽然现代 Python 推荐使用 asyncio.run() 自动管理循环，但理解底层机制有助于调试复杂问题。

import asyncio

loop = asyncio.get_event_loop()

async def worker():
    print('Worker running')
    await asyncio.sleep(1)

# 手动运行循环
loop.run_until_complete(worker())
loop.close()

asyncio 核心库详解

asyncio 提供了丰富的工具函数来简化开发：

• asyncio.run()：启动事件循环，运行顶层协程。
• asyncio.create_task()：创建后台任务。
• asyncio.gather()：并发执行多个协程并聚合结果。
• asyncio.sleep()：非阻塞延时。
• asyncio.Future：底层的未完成对象，用于回调链式处理。

高级用法中，结合 Future 可以实现更灵活的状态通知：

import asyncio

async def producer():
    return "Result"

async def consumer():
    future = asyncio.Future()
    task = asyncio.create_task(producer())
    task.add_done_callback(lambda t: future.set_result(t.result()))
    result = await future
    print(f'Got: {result}')

asyncio.run(consumer())

aiohttp 异步网络库

aiohttp 是 Python 生态中最流行的异步 HTTP 客户端和服务端框架。

安装依赖

pip install aiohttp

发送请求

使用 ClientSession 管理连接池，避免重复握手开销。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[:100])

asyncio.run(main())

POST 请求支持表单数据或 JSON 格式：

import aiohttp
import asyncio

async def post_json(session, url, data):
    async with session.post(url, json=data) as response:
        return await response.json()

async def main():
    async with aiohttp.ClientSession() as session:
        payload = {'name': 'Alice', 'age': 25}
        resp = await post_json(session, 'https://httpbin.org/post', payload)
        print(resp['json'])

asyncio.run(main())

实战案例：高并发 HTTP 客户端

假设我们需要同时抓取多个网站的状态，传统方式会串行等待，而异步方式可大幅缩短总耗时。

import aiohttp
import asyncio
import time

async def fetch_url(session, url):
    start = time.time()
    async with session.get(url) as resp:
        text = await resp.text()
        return url, len(text), time.time() - start

async def main():
    urls = [
        'https://www.python.org',
        'https://www.github.com',
        'https://www.google.com'
    ]
    
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        
        for url, length, duration in results:
            print(f'{url}: {length} bytes, {duration:.2f}s')

if __name__ == '__main__':
    total_start = time.time()
    asyncio.run(main())
    print(f'Total time: {time.time() - total_start:.2f}s')

实战案例：异步 Web 服务器

基于 aiohttp.web 可以快速搭建高性能 API 服务，无需引入 Gunicorn 等 WSGI 容器。

from aiohttp import web
import asyncio

async def handle_root(request):
    return web.Response(text='Hello, Async World!')

async def handle_api(request):
    data = {'status': 'ok', 'version': '1.0'}
    return web.json_response(data)

async def create_app():
    app = web.Application()
    app.router.add_get('/', handle_root)
    app.router.add_get('/api', handle_api)
    return app

if __name__ == '__main__':
    # 监听本地 8080 端口
    web.run_app(create_app(), host='localhost', port=8080)

总结

掌握 Python 异步编程的关键在于理解事件循环与协程的配合。通过 asyncio 和 aiohttp，开发者可以编写出资源占用低、吞吐量高的网络应用。建议在项目中遇到 I/O 瓶颈时优先考虑异步方案，并注意避免在协程中执行 CPU 密集型计算，以免阻塞事件循环。