python-async-patterns

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Python Async Patterns

Python异步模式

Asyncio patterns for concurrent Python programming.

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

Core Concepts

核心概念

python

import asyncio

python

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

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

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

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):
    """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

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!

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!

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
```
- 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

脚本

```
./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异步模式

Core Concepts

核心概念

Coroutine (must be awaited)

Coroutine (must be awaited)

Entry point

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

WRONG - blocks event loop

CORRECT

CORRECT

WRONG - orphaned task

WRONG - orphaned task

CORRECT - keep reference

CORRECT - keep reference

Quick Reference

快速参考

Async Context Manager

异步上下文管理器

Additional Resources

额外资源

Scripts

脚本

Assets

资源文件

See Also

另请参阅