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

任务调度与管理

在实际开发中，我们需要对任务进行更精细的控制，包括创建、取消和超时处理。

创建与取消任务

asyncio.create_task() 将协程封装为 Task 对象，便于管理。若需提前终止，可调用 cancel() 方法。

import asyncio

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

async def main():
    task = asyncio.create_task(task_worker())
    await asyncio.sleep(0.5)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print('Main: Task was cancelled')

if __name__ == '__main__':
    asyncio.run(main())

设置超时保护

防止某个任务无限期挂起，可使用 asyncio.wait_for() 设置时间限制。

import asyncio

async def slow_operation():
    await asyncio.sleep(2)
    return 'Done'

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

if __name__ == '__main__':
    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)

try:
    loop.run_until_complete(worker())
finally:
    loop.close()

注意：在 Python 3.7+ 中，asyncio.run() 会自动创建新的事件循环，通常无需手动获取或关闭。

核心库实战：aiohttp

aiohttp 是 Python 生态中最成熟的异步 HTTP 客户端与服务端库，完美配合 asyncio 使用。

安装依赖

pip install aiohttp

构建异步 HTTP 客户端

以下示例展示了如何并发发送多个请求并统计耗时。

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() - start_time
        return url, len(text), elapsed

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

if __name__ == '__main__':
    asyncio.run(main())

搭建异步 Web 服务

利用 aiohttp.web 模块可以快速构建高性能 API 服务。

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', 'message': 'Success'}
    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__':
    app = create_app()
    web.run_app(app, host='localhost', port=8080)

总结

掌握 Python 异步编程的关键在于理解协程的生命周期与事件循环的调度逻辑。通过 asyncio 基础组件与 aiohttp 网络库的结合，我们可以轻松构建高并发、低延迟的网络应用。建议在实际项目中多尝试混合使用同步与异步代码，并根据业务场景选择合适的并发策略。