Asyncio Programming

Asyncio 编程

概述

掌握使用 asyncio 进行 Python 异步编程。学习编写能够高效处理 I/O 密集型操作的并发代码、构建异步 Web 应用，并理解 async/await 范式。

学习目标

• 理解异步编程概念
• 使用 async/await 语法编写异步函数
• 使用 asyncio 管理并发操作
• 构建异步 Web 应用
• 高效处理异步 I/O 操作
• 调试和测试异步代码

核心主题

1. Async/Await 基础

• 理解协程
• async/await 语法
• 事件循环基础
• 运行异步函数
• 异步执行与同步执行的区别
• 常见陷阱

代码示例：

import asyncio
import time

# Synchronous version (slow)
def fetch_data_sync(url):
    print(f"Fetching {url}...")
    time.sleep(2)  # Simulating network delay
    return f"Data from {url}"

def main_sync():
    urls = ['url1', 'url2', 'url3']
    results = []
    for url in urls:
        data = fetch_data_sync(url)
        results.append(data)
    return results

# Takes 6 seconds (2 * 3)
start = time.time()
main_sync()
print(f"Sync took: {time.time() - start:.2f}s")

# Asynchronous version (fast)
async def fetch_data_async(url):
    print(f"Fetching {url}...")
    await asyncio.sleep(2)  # Non-blocking sleep
    return f"Data from {url}"

async def main_async():
    urls = ['url1', 'url2', 'url3']
    # Create tasks and run concurrently
    tasks = [fetch_data_async(url) for url in urls]
    results = await asyncio.gather(*tasks)
    return results

# Takes 2 seconds (concurrent execution)
start = time.time()
asyncio.run(main_async())
print(f"Async took: {time.time() - start:.2f}s")

2. Asyncio 任务与协程

• 创建和管理任务
• asyncio.gather() 与 asyncio.wait() 的区别
• 取消任务
• 任务组（Python 3.11+）
• 任务中的异常处理
• 超时

代码示例：

import asyncio

async def process_item(item_id, delay):
    print(f"Processing item {item_id}")
    await asyncio.sleep(delay)
    if item_id == 3:
        raise ValueError(f"Item {item_id} failed!")
    return f"Result {item_id}"

async def main():
    # Method 1: gather (returns results in order)
    tasks = [
        process_item(1, 1),
        process_item(2, 2),
        process_item(3, 1),
    ]
    try:
        results = await asyncio.gather(*tasks, return_exceptions=True)
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                print(f"Task {i} failed: {result}")
            else:
                print(f"Task {i} result: {result}")
    except Exception as e:
        print(f"Error: {e}")

    # Method 2: wait (returns done/pending sets)
    tasks = [
        asyncio.create_task(process_item(i, i))
        for i in range(1, 4)
    ]
    done, pending = await asyncio.wait(tasks, timeout=2.5)

    print(f"Completed: {len(done)}, Pending: {len(pending)}")

    # Cancel pending tasks
    for task in pending:
        task.cancel()

    # Method 3: Task groups (Python 3.11+)
    async with asyncio.TaskGroup() as tg:
        for i in range(1, 4):
            tg.create_task(process_item(i, 1))
    # All tasks completed or exception raised

asyncio.run(main())

3. 异步 I/O 操作

• 异步文件操作（aiofiles）
• 异步 HTTP 请求（aiohttp）
• 异步数据库操作（asyncpg、motor）
• 异步消息传递（aio-pika）
• 流与协议

代码示例：

import asyncio
import aiohttp
import aiofiles
from typing import List

# Async HTTP requests
async def fetch_url(session, url):
    async with session.get(url) as response:
        return await response.text()

async def fetch_multiple_urls(urls: List[str]):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        return results

# Async file operations
async def read_file_async(filepath):
    async with aiofiles.open(filepath, 'r') as f:
        content = await f.read()
        return content

async def write_file_async(filepath, content):
    async with aiofiles.open(filepath, 'w') as f:
        await f.write(content)

# Async database operations (example with asyncpg)
import asyncpg

async def fetch_users():
    conn = await asyncpg.connect(
        user='user',
        password='password',
        database='mydb',
        host='localhost'
    )
    try:
        rows = await conn.fetch('SELECT * FROM users')
        return rows
    finally:
        await conn.close()

# Usage
async def main():
    # Fetch URLs concurrently
    urls = [
        'https://api.example.com/data1',
        'https://api.example.com/data2',
        'https://api.example.com/data3',
    ]
    results = await fetch_multiple_urls(urls)

    # Read/write files
    content = await read_file_async('input.txt')
    await write_file_async('output.txt', content.upper())

    # Database operations
    users = await fetch_users()
    print(f"Found {len(users)} users")

asyncio.run(main())

4. 异步 Web 框架

• FastAPI 异步路由
• aiohttp Web 服务
• WebSocket 处理
• 后台任务
• 中间件与依赖项

代码示例：

# FastAPI async example
from fastapi import FastAPI, BackgroundTasks
import asyncio

app = FastAPI()

# Async route
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # Async database call
    user = await fetch_user_from_db(user_id)
    return user

# Background task
async def send_notification(email: str, message: str):
    await asyncio.sleep(2)  # Simulate email sending
    print(f"Sent email to {email}: {message}")

@app.post("/orders/")
async def create_order(order_data: dict, background_tasks: BackgroundTasks):
    # Process order synchronously
    order_id = save_order(order_data)

    # Send notification in background
    background_tasks.add_task(
        send_notification,
        order_data['customer_email'],
        f"Order #{order_id} created"
    )

    return {"order_id": order_id}

# aiohttp web server
from aiohttp import web

async def handle_request(request):
    name = request.match_info.get('name', 'Anonymous')
    await asyncio.sleep(1)  # Async operation
    return web.json_response({'message': f'Hello {name}'})

app = web.Application()
app.add_routes([web.get('/{name}', handle_request)])

# WebSocket example
async def websocket_handler(request):
    ws = web.WebSocketResponse()
    await ws.prepare(request)

    async for msg in ws:
        if msg.type == web.WSMsgType.TEXT:
            await ws.send_str(f"Echo: {msg.data}")
        elif msg.type == web.WSMsgType.ERROR:
            print(f'Error: {ws.exception()}')

    return ws

app.add_routes([web.get('/ws', websocket_handler)])

动手实践

项目 1：异步 Web 爬虫

构建带有速率限制的并发 Web 爬虫。

要求：

• 并发抓取多个网站
• 实现速率限制
• 妥善处理错误
• 将结果保存到异步数据库
• 跟踪进度
• 重试失败的请求

关键技能： aiohttp、异步 I/O、错误处理

项目 2：实时聊天服务

创建基于 WebSocket 的聊天应用。

要求：

• 使用 aiohttp 构建 WebSocket 服务
• 多个聊天室
• 用户身份验证
• 消息广播
• 连接管理
• 持久化消息历史记录

关键技能： WebSocket、异步服务、状态管理

项目 3：异步任务队列

构建分布式任务处理系统。

要求：

• 使用 Redis/RabbitMQ 构建任务队列
• 工作进程池管理
• 任务优先级管理
• 结果缓存
• 进度监控
• 优雅关闭

关键技能： 消息队列、并发工作进程、清理

评估标准

• 理解 async/await 语义
• 编写高效的并发代码
• 妥善处理异步异常
• 有效使用 asyncio 任务
• 构建异步 Web 应用
• 调试异步代码
• 管理异步资源（清理）

资源

官方文档

• asyncio 文档 - 官方文档
• aiohttp - 异步 HTTP 客户端/服务端
• FastAPI - 现代异步 Web 框架

学习平台

• 在 Python 中使用 Asyncio - Real Python 指南
• Python 并发编程 - O'Reilly 图书
• Asyncio 实用示例 - 实用示例

工具

• aiohttp - 异步 HTTP
• aiofiles - 异步文件 I/O
• asyncpg - 异步 PostgreSQL
• aiodebug - 异步调试

后续步骤

掌握 asyncio 后，可以继续探索：

• Multiprocessing - CPU 密集型并行处理
• Celery - 分布式任务队列
• gRPC - 异步 RPC 框架
• Kafka - 异步事件流