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

asyncio 库进阶用法

除了基础调度，asyncio 还提供了丰富的工具类来处理复杂场景，如 Future 对象回调、自定义事件循环等。

import asyncio

async def future_example():
    future = asyncio.Future()
    
    async def producer():
        await asyncio.sleep(1)
        future.set_result('Done')
    
    asyncio.create_task(producer())
    result = await future
    print(f'Result: {result}')

asyncio.run(future_example())

aiohttp 网络通信实战

安装与基础请求

aiohttp 是 Python 生态中最流行的异步 HTTP 客户端库。安装后，利用 ClientSession 管理连接池。

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 载荷同样简单，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': '张三', 'age': 25}
        resp = await post_json(session, 'https://httpbin.org/post', payload)
        print(resp)

asyncio.run(main())

实战案例：高性能异步服务

异步 HTTP 客户端

下面是一个批量抓取网页的完整示例，展示了如何利用并发加速 I/O 密集型任务。

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: {length}, Time: {elapsed_time:.2f}s')

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

异步 Web 服务器

基于 aiohttp.web 模块可以快速搭建轻量级 API 服务，支持静态文件托管。

from aiohttp import web
import asyncio

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

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

async def create_app():
    app = web.Application()
    app.add_routes([
        web.get('/', handle_root),
        web.get('/api', handle_api),
        web.static('/static', './static')
    ])
    return app

if __name__ == '__main__':
    # 启动服务监听 localhost:8080
    asyncio.run(web.run_app(create_app(), host='localhost', port=8080))

结语

掌握异步编程的关键在于理解事件循环的工作机制以及何时该使用 await。通过 asyncio 和 aiohttp，我们可以用较少的资源构建出高并发的网络应用。建议在实际项目中多尝试混合模式，根据业务特性选择同步或异步方案，以达到最佳性能平衡。