async-python-patterns

异步 Python 模式

使用 asyncio、并发编程模式和 async/await 实现异步 Python 应用的综合指南，助你构建高性能、非阻塞系统。

何时使用本技能

• 构建异步 Web API（FastAPI、aiohttp、Sanic）
• 实现并发 I/O 操作（数据库、文件、网络）
• 创建带并发请求的网页爬虫
• 开发实时应用（WebSocket 服务、聊天系统）
• 同时处理多个独立任务
• 构建采用异步通信的微服务
• 优化 I/O 密集型工作负载
• 实现异步后台任务与队列

同步 vs 异步决策指南

在采用异步之前，请先考虑它是否适合你的用例。

用例	推荐方案
大量并发的网络/数据库调用	asyncio
CPU 密集型计算	multiprocessing 或线程池
I/O 与 CPU 混合	用 asyncio.to_thread() 卸载 CPU 工作
简单脚本、连接数少	同步（更简单，更易调试）
高并发的 Web API	异步框架（FastAPI、aiohttp）

关键规则： 在同一条调用路径上，要么完全同步，要么完全异步。混用会带来隐蔽的阻塞和复杂性。

核心概念

1. 事件循环

事件循环是 asyncio 的核心，负责管理和调度异步任务。

关键特性：

• 单线程协作式多任务
• 调度协程执行
• 以非阻塞方式处理 I/O 操作
• 管理回调和 future

2. 协程

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

语法：

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

3. 任务

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

4. Future

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

5. 异步上下文管理器

支持 async with 的资源，可进行正确的清理。

6. 异步迭代器

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

快速开始

import asyncio

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

# Python 3.7+
asyncio.run(main())

基础模式

模式 1：基本 async/await

import asyncio

async def fetch_data(url: str) -> dict:
    """Fetch data from URL asynchronously."""
    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())

模式 2：使用 gather() 并发执行

import asyncio
from typing import List

async def fetch_user(user_id: int) -> dict:
    """Fetch user data."""
    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]:
    """Fetch multiple users concurrently."""
    tasks = [fetch_user(uid) for uid in user_ids]
    results = await asyncio.gather(*tasks)
    return results

async def main():
    user_ids = [1, 2, 3, 4, 5]
    users = await fetch_all_users(user_ids)
    print(f"Fetched {len(users)} users")

asyncio.run(main())

模式 3：任务创建与管理

import asyncio

async def background_task(name: str, delay: int):
    """Long-running background task."""
    print(f"{name} started")
    await asyncio.sleep(delay)
    print(f"{name} completed")
    return f"Result from {name}"

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

    # Do other work
    print("Main: doing other work")
    await asyncio.sleep(0.5)

    # Wait for tasks
    result1 = await task1
    result2 = await task2

    print(f"Results: {result1}, {result2}")

asyncio.run(main())

模式 4：异步代码中的错误处理

import asyncio
from typing import List, Optional

async def risky_operation(item_id: int) -> dict:
    """Operation that might fail."""
    await asyncio.sleep(0.1)
    if item_id % 3 == 0:
        raise ValueError(f"Item {item_id} failed")
    return {"id": item_id, "status": "success"}

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

async def process_items(item_ids: List[int]):
    """Process multiple items with error handling."""
    tasks = [safe_operation(iid) for iid in item_ids]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    # Filter out failures
    successful = [r for r in results if r is not None and not isinstance(r, Exception)]
    failed = [r for r in results if r is None or isinstance(r, Exception)]

    print(f"Success: {len(successful)}, Failed: {len(failed)}")
    return successful

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

模式 5：超时处理

import asyncio

async def slow_operation(delay: int) -> str:
    """Operation that takes time."""
    await asyncio.sleep(delay)
    return f"Completed after {delay}s"

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

asyncio.run(with_timeout())

详细示例与模式

详细章节（以 ## Advanced Patterns 开头）位于 references/details.md。当上面的导航摘要不够用时，请阅读该文件。

常见陷阱

1. 忘记 await

# Wrong - returns coroutine object, doesn't execute
result = async_function()

# Correct
result = await async_function()

2. 阻塞事件循环

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

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

3. 未处理取消

async def cancelable_task():
    """Task that handles cancellation."""
    try:
        while True:
            await asyncio.sleep(1)
            print("Working...")
    except asyncio.CancelledError:
        print("Task cancelled, cleaning up...")
        # Perform cleanup
        raise  # Re-raise to propagate cancellation

4. 混用同步与异步代码

# Wrong - can't call async from sync directly
def sync_function():
    result = await async_function()  # SyntaxError!

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

测试异步代码

import asyncio
import pytest

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

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