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

异步编程核心概念

在构建高并发应用时，理解 Python 的异步机制至关重要。它允许程序在等待 I/O 操作完成时继续执行其他任务，从而显著提升效率。

什么是异步编程

异步编程通过非阻塞操作提高执行效率。当程序遇到网络请求或文件读写等耗时操作时，无需挂起整个线程，而是切换到其他任务。这种方式减少了资源消耗，简化了代码结构。

常见应用场景包括网络通信（HTTP、WebSocket）、大文件操作以及数据库查询。

协程的定义与使用

协程（Coroutine）是轻量级的线程，支持暂停和恢复。在 Python 中，使用 async def 关键字定义协程。

import asyncio

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

asyncio.run(hello())

这里我们直接调用 asyncio.run() 启动事件循环并运行协程。注意 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.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 设置超时时间：

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

asyncio 库核心函数

asyncio 是标准库中的核心模块，提供了事件循环和任务调度功能。

• asyncio.run()：入口点，自动创建并关闭事件循环。
• asyncio.create_task()：创建后台任务。
• asyncio.gather()：并发执行多个协程并等待结果。
• asyncio.wait_for()：带超时的等待。
• asyncio.sleep()：模拟异步休眠。

对于更复杂的状态管理，还可以结合 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 异步网络库

处理 HTTP 请求时，推荐使用 aiohttp 库，它专为异步设计。

安装与基础用法

首先安装依赖：

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 请求时，可以直接传递数据字典：

import aiohttp
import asyncio

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 客户端

让我们构建一个能并发抓取多个网站信息的客户端。相比同步方式，这能大幅减少总耗时。

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'
    ]
    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 服务。

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))

这里定义了根路径和 API 接口路由。启动后，你可以访问 http://localhost:8080/api 获取 JSON 数据。这种基于协程的服务器模型在处理大量连接时表现优异。

总结

掌握异步编程的关键在于理解事件循环和协程的生命周期。通过 asyncio 和 aiohttp，我们可以编写出既高效又简洁的网络应用。建议在实际项目中多尝试并发场景，体会非阻塞 IO 带来的性能提升。