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

任务控制：取消与超时

在实际开发中，我们可能需要取消某个任务或限制其执行时间。

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 slow_task():
    print('Starting...')
    await asyncio.sleep(2)
    print('Done!')

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

asyncio.run(main())

asyncio 库深度解析

asyncio 是 Python 标准库中异步编程的核心。除了基础的任务管理，它还提供了更高级的用法。

常用函数速查

• asyncio.run()：入口点，用于运行顶层协程。
• asyncio.create_task()：将协程包装为任务对象。
• asyncio.gather()：收集多个任务的结果。
• asyncio.sleep()：非阻塞地暂停指定秒数。

使用 Future 对象

有时候我们需要手动控制任务的完成状态，这时可以借助 Future 对象。

import asyncio

async def hello():
    return 'Result from hello'

async def main():
    future = asyncio.Future()
    task = asyncio.create_task(hello())
    
    # 当任务完成时设置 Future 结果
    task.add_done_callback(lambda t: future.set_result(t.result()))
    
    result = await future
    print(result)

asyncio.run(main())

aiohttp 网络请求实战

对于网络应用，aiohttp 是最流行的异步 HTTP 客户端和服务器库。

安装与基础 GET 请求

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 数据

发送表单数据和 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:
        payload = {'name': '张三', 'age': 25}
        response = await post_json(session, 'https://httpbin.org/post', payload)
        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',
        '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())
    total_time = time.time() - start_time
    print(f'总耗时：{total_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 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))

通过上述实践，你可以看到异步模型如何在保持代码简洁的同时，显著提升 I/O 密集型应用的吞吐量。在实际项目中，合理运用 asyncio 和 aiohttp，能让你的 Python 应用更好地应对高并发挑战。