---

name: python-async-ops description: "Python asyncio 并发编程模式。" license: MIT allowed-tools: "Read Write" metadata: author: claude-mods

Python 异步模式

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

核心概念

import asyncio

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

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]:
    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):
    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):
    async with asyncio.TaskGroup() as tg:
        for item in items:
            tg.create_task(process_one(item))

模式 4：超时

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

关键警告

# 错误 - 阻塞事件循环
async def bad():
    time.sleep(5)
    requests.get(url)

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

快速参考

| 模式 | 使用场景 | |---------|----------| | 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 - 队列、锁、生产者-消费者
• ./references/aiohttp-patterns.md - HTTP 客户端/服务器模式
• ./references/production-patterns.md - 优雅关闭、健康检查

另请参阅

• python-fastapi-ops - 异步 Web API
• python-observability-ops - 异步日志和追踪