python-async-ops

Python 异步模式

用于并发 Python 编程的 asyncio 模式。

何时使用异步或同步

适用异步的情况	适用同步的情况
I/O 密集型操作（HTTP、DB、文件）	CPU 密集型计算
高并发（数百个以上连接）	简单脚本、一次性任务
WebSocket/流式连接	小规模数据处理
存在网络调用的微服务	单个顺序操作

决策流程：

1. 是否为 CPU 密集型？→ 同步（或多进程）
2. 是否为高并发的 I/O 密集型？→ 异步
3. 是否为连接数较少的简单 I/O？→ 使用同步即可

核心概念

import asyncio

# Coroutine (must be awaited)
async def fetch(url: str) -> str:
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

# Entry point
async def main():
    result = await fetch("https://example.com")
    return result

asyncio.run(main())

模式 1：使用 gather 并发执行

async def fetch_all(urls: list[str]) -> list[str]:
    """Fetch multiple URLs concurrently."""
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_one(session, url) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

模式 2：限制并发

async def fetch_with_limit(urls: list[str], limit: int = 10):
    """Limit concurrent requests."""
    semaphore = asyncio.Semaphore(limit)

    async def bounded_fetch(url):
        async with semaphore:
            return await fetch_one(url)

    return await asyncio.gather(*[bounded_fetch(url) for url in urls])

模式 3：TaskGroup（Python 3.11+）

async def process_items(items):
    """Structured concurrency with automatic cleanup."""
    async with asyncio.TaskGroup() as tg:
        for item in items:
            tg.create_task(process_one(item))
    # All tasks complete here, or exception raised

模式 4：超时

async def with_timeout():
    try:
        async with asyncio.timeout(5.0):  # Python 3.11+
            result = await slow_operation()
    except asyncio.TimeoutError:
        result = None
    return result

重要警告

# WRONG - blocks event loop
async def bad():
    time.sleep(5)         # Never use time.sleep!
    requests.get(url)     # Blocking I/O!

# CORRECT
async def good():
    await asyncio.sleep(5)
    async with aiohttp.ClientSession() as s:
        await s.get(url)

# WRONG - orphaned task
async def bad():
    asyncio.create_task(work())  # May be garbage collected!

# CORRECT - keep reference
async def good():
    task = asyncio.create_task(work())
    await task

快速参考

模式	使用场景
gather(*tasks)	多个独立操作
Semaphore(n)	速率限制、资源约束
TaskGroup()	结构化并发（3.11+）
Queue()	生产者-消费者
timeout(s)	超时包装器（3.11+）
Lock()	共享可变状态

异步上下文管理器

from contextlib import asynccontextmanager

@asynccontextmanager
async def managed_connection():
    conn = await create_connection()
    try:
        yield conn
    finally:
        await conn.close()

更多资源

如需了解详细模式，请加载：

• ./references/concurrency-patterns.md - Queue、Lock、生产者-消费者
• ./references/aiohttp-patterns.md - HTTP 客户端/服务端模式
• ./references/mixing-sync-async.md - run_in_executor、线程池
• ./references/debugging-async.md - 调试模式、性能分析、问题排查
• ./references/production-patterns.md - 优雅关闭、健康检查、信号处理
• ./references/error-handling.md - 带退避的重试、熔断器、部分故障
• ./references/performance.md - uvloop、连接池、缓冲区大小设置

脚本

• ./scripts/find-blocking-calls.sh - 扫描代码，查找异步函数中的阻塞调用

资源文件

• ./assets/async-project-template.py - 可用于生产环境的异步应用骨架

---

另请参阅

前置条件：

• python-typing-ops - 异步函数的类型提示

相关技能：

• python-fastapi-ops - 异步 Web API
• python-observability-ops - 异步日志记录与追踪
• python-database-ops - 异步数据库访问