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

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

相比顺序执行，这种方式能大幅缩短总耗时，因为三个任务在等待 sleep 期间都在排队，而不是串行等待。

任务取消与超时

有时候我们需要控制任务的执行时间或主动终止某个任务。

取消任务：

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

设置超时：

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

wait_for 会在指定时间内未完成则抛出异常，非常适合防止服务长时间无响应。

实战：使用 aiohttp 进行网络交互

aiohttp 是 Python 中最流行的异步 HTTP 客户端和服务端库，完美契合 asyncio 生态。

安装与基础请求

首先确保安装依赖：

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 payload 也非常简单：

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

注意 json= 参数会自动序列化字典并设置正确的 Content-Type，比手动传 data 更方便。

综合案例：高性能异步 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())
    elapsed_time = time.time() - start_time
    print(f'总耗时：{elapsed_time:.2f}秒')

在这个案例中，所有请求几乎同时发出，总耗时远小于逐个请求的时间之和，体现了异步并发的高吞吐能力。

构建异步 Web 服务器

除了客户端，aiohttp 也能轻松搭建服务端。

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

路由定义清晰，中间件支持完善，非常适合构建微服务接口。

总结

掌握 Python 异步编程的关键在于理解事件循环的工作机制以及 asyncio 库提供的调度工具。通过合理使用 gather、create_task 和 wait_for，可以编写出既高效又健壮的并发应用。无论是高频网络请求还是实时数据处理，异步模型都是提升系统性能的首选方案。建议在实际项目中多尝试重构同步代码为异步逻辑，逐步积累手感。