python-async-patterns

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Python Async Patterns

Python Asyncio 并发编程模式

Asyncio patterns for concurrent Python programming.

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

Core Concepts

核心概念

python

import asyncio

python

import asyncio

Coroutine (must be awaited)

协程（必须使用await调用）

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())

undefined

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

asyncio.run(main())

undefined

Pattern 1: Concurrent with gather

模式1：使用gather实现并发

python

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)

python

async def fetch_all(urls: list[str]) -> list[str]:
    """并发获取多个URL内容。"""
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_one(session, url) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

Pattern 2: Bounded Concurrency

模式2：有界并发

python

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])

python

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])

Pattern 3: TaskGroup (Python 3.11+)

模式3：TaskGroup（Python 3.11+）

python

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

python

async def process_items(items):
    """带有自动清理的结构化并发。"""
    async with asyncio.TaskGroup() as tg:
        for item in items:
            tg.create_task(process_one(item))
    # 所有任务在此处完成，或抛出异常

Pattern 4: Timeout

模式4：超时处理

python

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

python

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

Critical Warnings

重要注意事项

python

undefined

python

undefined

WRONG - blocks event loop

错误示例 - 阻塞事件循环

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

async def bad(): time.sleep(5) # 绝不要使用time.sleep！ requests.get(url) # 阻塞式I/O操作！

CORRECT

正确示例

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


```python

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


```python

WRONG - orphaned task

错误示例 - 孤立任务

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

async def bad(): asyncio.create_task(work()) # 可能会被垃圾回收！

CORRECT - keep reference

正确示例 - 保留引用

async def good(): task = asyncio.create_task(work()) await task

undefined

async def good(): task = asyncio.create_task(work()) await task

undefined

Quick Reference

速查表

Pattern	Use Case
`gather(*tasks)`	Multiple independent operations
`Semaphore(n)`	Rate limiting, resource constraints
`TaskGroup()`	Structured concurrency (3.11+)
`Queue()`	Producer-consumer
`timeout(s)`	Timeout wrapper (3.11+)
`Lock()`	Shared mutable state

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

Async Context Manager

异步上下文管理器

python

from contextlib import asynccontextmanager

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

python

from contextlib import asynccontextmanager

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

Additional Resources

额外资源

For detailed patterns, load:

```
./references/concurrency-patterns.md
```
- Queue, Lock, producer-consumer
```
./references/aiohttp-patterns.md
```
- HTTP client/server patterns
```
./references/mixing-sync-async.md
```
- run_in_executor, thread pools
```
./references/debugging-async.md
```
- Debug mode, profiling, finding issues
```
./references/production-patterns.md
```
- Graceful shutdown, health checks, signal handling
```
./references/error-handling.md
```
- Retry with backoff, circuit breakers, partial failures
```
./references/performance.md
```
- uvloop, connection pooling, buffer sizing

如需详细模式说明，请查看：

```
./references/concurrency-patterns.md
```
- 队列、锁、生产者-消费者模型
```
./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

脚本工具

```
./scripts/find-blocking-calls.sh
```
- Scan code for blocking calls in async functions

```
./scripts/find-blocking-calls.sh
```
- 扫描异步函数中的阻塞调用

Assets

资源文件

```
./assets/async-project-template.py
```
- Production-ready async app skeleton

```
./assets/async-project-template.py
```
- 生产级异步应用骨架

另请参阅

Prerequisites:

```
python-typing-patterns
```
- Type hints for async functions

Related Skills:

```
python-fastapi-patterns
```
- Async web APIs
```
python-observability-patterns
```
- Async logging and tracing
```
python-database-patterns
```
- Async database access

前置要求：

```
python-typing-patterns
```
- 异步函数的类型提示

相关技能：

```
python-fastapi-patterns
```
- 异步Web API
```
python-observability-patterns
```
- 异步日志与追踪
```
python-database-patterns
```
- 异步数据库访问

python-async-patterns

Original

Translation

Python Async Patterns

Python Asyncio 并发编程模式

Core Concepts

核心概念

Coroutine (must be awaited)

协程（必须使用await调用）

Entry point

入口函数

Pattern 1: Concurrent with gather

模式1：使用gather实现并发

Pattern 2: Bounded Concurrency

模式2：有界并发

Pattern 3: TaskGroup (Python 3.11+)

模式3：TaskGroup（Python 3.11+）

Pattern 4: Timeout

模式4：超时处理

Critical Warnings

重要注意事项

WRONG - blocks event loop

错误示例 - 阻塞事件循环

CORRECT

正确示例

WRONG - orphaned task

错误示例 - 孤立任务

CORRECT - keep reference

正确示例 - 保留引用

Quick Reference

速查表

Async Context Manager

异步上下文管理器

Additional Resources

额外资源

Scripts

脚本工具

Assets

资源文件

See Also

另请参阅