async-python-patterns

异步 Python 模式

使用 asyncio、并发编程模式和 async/await 构建异步 Python 应用，打造高性能、非阻塞系统。

适用场景

• 构建异步 Web API（FastAPI、aiohttp、Sanic）
• 实现并发 I/O 操作（数据库、文件、网络）
• 创建并发请求的网页抓取工具
• 开发实时应用（WebSocket 服务、聊天系统）
• 同时处理多个独立任务
• 构建采用异步通信的微服务
• 优化受 I/O 限制的工作负载
• 实现异步后台任务和队列
• 在多线程、多进程和 asyncio 之间进行选择

核心概念

事件循环

• 单线程协作式多任务处理
• 调度协程执行
• 处理 I/O 操作而不造成阻塞
• 管理回调和 future

协程

使用 async def 定义、可以暂停和恢复的函数。

async def my_coroutine():
    result = await some_async_operation()
    return result

任务

在事件循环上并发运行的已调度协程。

Future

表示异步操作最终结果的底层对象。

异步上下文管理器

支持使用 async with 进行适当清理的资源。

异步迭代器

支持使用 async for 遍历异步数据源的对象。

快速入门

import asyncio

async def main():
    print("Hello")
    await asyncio.sleep(1)
    print("World")

asyncio.run(main())

基本模式

基本的 Async/Await

import asyncio

async def fetch_data(url: str) -> dict:
    await asyncio.sleep(1)  # Simulate I/O
    return {"url": url, "data": "result"}

async def main():
    result = await fetch_data("https://api.example.com")
    print(result)

asyncio.run(main())

使用 gather() 并发执行

import asyncio
from typing import List

async def fetch_user(user_id: int) -> dict:
    await asyncio.sleep(0.5)
    return {"id": user_id, "name": f"User {user_id}"}

async def fetch_all_users(user_ids: List[int]) -> List[dict]:
    tasks = [fetch_user(uid) for uid in user_ids]
    return await asyncio.gather(*tasks)

asyncio.run(fetch_all_users([1, 2, 3, 4, 5]))

任务的创建与管理

import asyncio

async def background_task(name: str, delay: int):
    await asyncio.sleep(delay)
    return f"Result from {name}"

async def main():
    task1 = asyncio.create_task(background_task("Task 1", 2))
    task2 = asyncio.create_task(background_task("Task 2", 1))

    # Do other work while tasks run
    await asyncio.sleep(0.5)

    result1 = await task1
    result2 = await task2
    print(f"Results: {result1}, {result2}")

asyncio.run(main())

错误处理

import asyncio
from typing import Optional

async def safe_operation(item_id: int) -> Optional[dict]:
    try:
        return await risky_operation(item_id)
    except ValueError as e:
        print(f"Error: {e}")
        return None

async def process_items(item_ids):
    tasks = [safe_operation(iid) for iid in item_ids]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    successful = [r for r in results if r is not None and not isinstance(r, Exception)]
    failed = [r for r in results if isinstance(r, Exception)]
    return successful

超时处理

import asyncio

async def with_timeout():
    try:
        result = await asyncio.wait_for(slow_operation(5), timeout=2.0)
    except asyncio.TimeoutError:
        print("Operation timed out")

使用 Semaphore 进行速率限制

import asyncio
from typing import List

async def api_call(url: str, semaphore: asyncio.Semaphore) -> dict:
    async with semaphore:
        await asyncio.sleep(0.5)  # Simulate API call
        return {"url": url, "status": 200}

async def rate_limited_requests(urls: List[str], max_concurrent: int = 5):
    semaphore = asyncio.Semaphore(max_concurrent)
    tasks = [api_call(url, semaphore) for url in urls]
    return await asyncio.gather(*tasks)

选择指南

适合使用 asyncio 的场景

• 受 I/O 限制的任务（网络、数据库、文件）
• 需要处理大量并发连接
• Web 服务器和 API 客户端
• WebSocket 应用

适合使用 threading 的场景

• 使用现有同步库处理受 I/O 限制的任务
• 需要简单的并行处理
• 与会释放 GIL 的 C 扩展交互

适合使用 multiprocessing 的场景

• 受 CPU 限制的任务（计算、数据处理）
• 需要真正的并行处理（绕过 GIL）
• 无需共享内存的任务

常见陷阱

忘记使用 await

# Wrong - returns coroutine object
result = async_function()

# Correct
result = await async_function()

阻塞事件循环

# Wrong - blocks event loop
import time
async def bad():
    time.sleep(1)  # Blocks!

# Correct
async def good():
    await asyncio.sleep(1)  # Non-blocking

未处理取消操作

async def cancelable_task():
    try:
        while True:
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        # Perform cleanup
        raise  # Re-raise to propagate

混合同步和异步代码

# Wrong
def sync_function():
    result = await async_function()  # SyntaxError!

# Correct
def sync_function():
    result = asyncio.run(async_function())

测试异步代码

import asyncio
import pytest

@pytest.mark.asyncio
async def test_async_function():
    result = await fetch_data("https://api.example.com")
    assert result is not None

@pytest.mark.asyncio
async def test_with_timeout():
    with pytest.raises(asyncio.TimeoutError):
        await asyncio.wait_for(slow_operation(5), timeout=1.0)

最佳实践

1. 使用 asyncio.run() 作为入口点（Python 3.7+）
2. 始终等待协程以执行它们
3. 使用 gather() 并发执行多个任务
4. 使用 try/except 妥善处理错误
5. 使用超时防止操作挂起
6. 使用连接池提升性能
7. 避免在异步代码中执行阻塞操作——使用 run_in_executor
8. 使用信号量进行速率限制
9. 妥善处理任务取消
10. 使用 pytest-asyncio 测试异步代码

参考资料

• references/async-patterns.md - 异步上下文管理器、异步迭代器/生成器、生产者-消费者模式、异步锁与同步机制、使用 aiohttp 进行网页抓取、异步数据库操作、WebSocket 服务实现、连接池、批处理操作、在执行器中运行阻塞操作

资源

• Python asyncio 文档：https://docs.python.org/3/library/asyncio.html
• aiohttp：异步 HTTP 客户端/服务端
• FastAPI：现代异步 Web 框架
• asyncpg：异步 PostgreSQL 驱动程序
• motor：异步 MongoDB 驱动程序