asyncio-concurrency-patterns

Asyncio 并发模式

全面掌握 Python asyncio 库和并发编程模式的技能。本技能涵盖事件循环、协程、任务、Future、同步原语、异步上下文管理器，以及用于构建高性能异步应用的生产级模式。

何时使用此技能

在以下情况下使用此技能：

• 构建需要处理大量并发操作的 I/O 密集型应用
• 创建 Web 服务器、API 客户端或 WebSocket 应用
• 使用事件驱动架构实现实时系统
• 通过并发处理请求来优化应用性能
• 协调多个异步操作，并妥善处理错误
• 构建后台任务处理器或作业队列
• 实现异步数据库操作和连接池
• 创建聊天应用、实时仪表盘或通知系统
• 高效处理并行 HTTP 请求
• 管理由多个事件源驱动的 WebSocket 连接
• 使用异步通信模式构建微服务
• 优化网络应用中的资源利用率

核心概念

什么是 Asyncio？

Asyncio 是 Python 内置的库，可使用 async/await 语法编写并发代码。它提供：

• 事件循环：asyncio 的核心，用于调度和运行异步任务
• 协程：使用 async def 定义、可暂停和恢复的函数
• 任务：已调度并发运行的协程
• Future：表示异步操作结果的底层对象
• 同步原语：用于协调的锁、信号量和事件

事件循环基础

事件循环是 asyncio 的核心执行机制：

import asyncio

# Get or create an event loop
loop = asyncio.get_event_loop()

# Run a coroutine until complete
loop.run_until_complete(my_coroutine())

# Modern approach (Python 3.7+)
asyncio.run(my_coroutine())

事件循环的关键概念：

1. 单线程并发：一个线程，多个任务
2. 协作式多任务处理：任务主动让出控制权
3. I/O 多路复用：高效处理多个 I/O 操作
4. 非阻塞操作：无需等待 I/O，可执行其他工作

协程与函数

常规函数：

def fetch_data():
    # Blocks until complete
    return requests.get('http://api.example.com')

协程：

async def fetch_data():
    # Yields control while waiting
    async with aiohttp.ClientSession() as session:
        async with session.get('http://api.example.com') as resp:
            return await resp.text()

任务与 Future

任务会包装协程，并将其调度到事件循环中：

# Create a task
task = asyncio.create_task(my_coroutine())

# Task runs in background
# ... do other work ...

# Wait for result
result = await task

Future 表示最终会产生的结果：

# Low-level future (rarely used directly)
future = asyncio.Future()

# Set result
future.set_result(42)

# Get result
result = await future

异步上下文管理器

使用异步设置和清理来管理资源：

class AsyncResource:
    async def __aenter__(self):
        # Async setup
        await self.connect()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        # Async cleanup
        await self.disconnect()

# Usage
async with AsyncResource() as resource:
    await resource.do_work()

并发模式

模式 1：Gather——并发执行

并发运行多个协程，并等待它们全部完成：

import asyncio
import aiohttp

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    async with aiohttp.ClientSession() as session:
        # Run all fetches concurrently
        results = await asyncio.gather(
            fetch(session, 'http://python.org'),
            fetch(session, 'http://docs.python.org'),
            fetch(session, 'http://pypi.org')
        )
        return results

# Results is a list in the same order as inputs
results = asyncio.run(main())

适用场景：

• 需要所有结果
• 顺序很重要
• 希望在首次出现异常时快速失败（默认行为）
• 可以通过 return_exceptions=True 处理部分结果

模式 2：Wait——灵活等待

更灵活地控制多个任务的等待方式：

import asyncio

async def task_a():
    await asyncio.sleep(2)
    return 'A'

async def task_b():
    await asyncio.sleep(1)
    return 'B'

async def main():
    tasks = [
        asyncio.create_task(task_a()),
        asyncio.create_task(task_b())
    ]

    # Wait for first to complete
    done, pending = await asyncio.wait(
        tasks,
        return_when=asyncio.FIRST_COMPLETED
    )

    # Get first result
    first_result = done.pop().result()

    # Cancel remaining
    for task in pending:
        task.cancel()

    return first_result

result = asyncio.run(main())  # Returns 'B' after 1 second

等待策略：

• FIRST_COMPLETED：首个任务完成时返回
• FIRST_EXCEPTION：首个任务引发异常时返回
• ALL_COMPLETED：等待所有任务（默认）

模式 3：Semaphore——限制并发

控制并发操作的最大数量：

import asyncio
import aiohttp

async def fetch_with_limit(session, url, semaphore):
    async with semaphore:
        # Only N requests run concurrently
        async with session.get(url) as resp:
            return await resp.text()

async def main():
    # Limit to 5 concurrent requests
    semaphore = asyncio.Semaphore(5)

    urls = [f'http://api.example.com/item/{i}' for i in range(100)]

    async with aiohttp.ClientSession() as session:
        tasks = [
            fetch_with_limit(session, url, semaphore)
            for url in urls
        ]
        results = await asyncio.gather(*tasks)

    return results

asyncio.run(main())

适用场景：

• 限制 API 请求速率
• 控制数据库连接的使用
• 防止资源耗尽
• 遵守外部服务的限制

模式 4：Lock——互斥

确保同一时间只有一个协程访问某个资源：

import asyncio

class SharedCounter:
    def __init__(self):
        self.value = 0
        self.lock = asyncio.Lock()

    async def increment(self):
        async with self.lock:
            # Critical section - only one coroutine at a time
            current = self.value
            await asyncio.sleep(0)  # Simulate async work
            self.value = current + 1

async def worker(counter):
    for _ in range(100):
        await counter.increment()

async def main():
    counter = SharedCounter()

    # Run 10 workers concurrently
    await asyncio.gather(*[worker(counter) for _ in range(10)])

    print(f"Final count: {counter.value}")  # Always 1000

asyncio.run(main())

模式 5：Event——信号通知

使用事件协调多个协程：

import asyncio

async def waiter(event, name):
    print(f'{name} waiting for event')
    await event.wait()
    print(f'{name} received event')

async def setter(event):
    await asyncio.sleep(2)
    print('Setting event')
    event.set()

async def main():
    event = asyncio.Event()

    # Multiple waiters
    await asyncio.gather(
        waiter(event, 'Waiter 1'),
        waiter(event, 'Waiter 2'),
        waiter(event, 'Waiter 3'),
        setter(event)
    )

asyncio.run(main())

模式 6：Queue——生产者/消费者

协调生产者与消费者之间的工作：

import asyncio

async def producer(queue, n):
    for i in range(n):
        await asyncio.sleep(0.1)
        await queue.put(f'item-{i}')
        print(f'Produced item-{i}')

    # Signal completion
    await queue.put(None)

async def consumer(queue, name):
    while True:
        item = await queue.get()

        if item is None:
            # Propagate sentinel to other consumers
            await queue.put(None)
            break

        print(f'{name} processing {item}')
        await asyncio.sleep(0.2)
        queue.task_done()

async def main():
    queue = asyncio.Queue()

    # Start producer and consumers
    await asyncio.gather(
        producer(queue, 10),
        consumer(queue, 'Consumer-1'),
        consumer(queue, 'Consumer-2'),
        consumer(queue, 'Consumer-3')
    )

asyncio.run(main())

任务管理

创建任务

基本任务创建：

import asyncio

async def background_task():
    await asyncio.sleep(10)
    return 'Done'

async def main():
    # Create task - starts running immediately
    task = asyncio.create_task(background_task())

    # Do other work while task runs
    await asyncio.sleep(1)

    # Wait for result
    result = await task
    return result

asyncio.run(main())

命名任务（Python 3.8+）：

task = asyncio.create_task(
    background_task(),
    name='my-background-task'
)

print(task.get_name())  # 'my-background-task'

取消任务

优雅取消：

import asyncio

async def long_running_task():
    try:
        while True:
            await asyncio.sleep(1)
            print('Working...')
    except asyncio.CancelledError:
        print('Task cancelled, cleaning up...')
        # Cleanup logic
        raise  # Re-raise to mark as cancelled

async def main():
    task = asyncio.create_task(long_running_task())

    # Let it run for 3 seconds
    await asyncio.sleep(3)

    # Request cancellation
    task.cancel()

    try:
        await task
    except asyncio.CancelledError:
        print('Task was cancelled')

asyncio.run(main())

使用上下文管理器取消：

import asyncio
from contextlib import suppress

async def run_with_timeout():
    task = asyncio.create_task(long_running_task())

    try:
        # Wait with timeout
        await asyncio.wait_for(task, timeout=5.0)
    except asyncio.TimeoutError:
        task.cancel()
        with suppress(asyncio.CancelledError):
            await task

任务中的异常处理

使用 Gather 处理异常：

import asyncio

async def failing_task(n):
    await asyncio.sleep(n)
    raise ValueError(f'Task {n} failed')

async def successful_task(n):
    await asyncio.sleep(n)
    return f'Task {n} succeeded'

async def main():
    # return_exceptions=True: Returns exceptions instead of raising
    results = await asyncio.gather(
        successful_task(1),
        failing_task(2),
        successful_task(3),
        return_exceptions=True
    )

    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f'Task {i} failed: {result}')
        else:
            print(f'Task {i} result: {result}')

asyncio.run(main())

获取任务异常：

import asyncio

async def main():
    task = asyncio.create_task(failing_task(1))

    # Wait for task
    await asyncio.sleep(2)

    # Check if task failed
    if task.done() and task.exception():
        print(f'Task failed with: {task.exception()}')

asyncio.run(main())

事件循环管理

事件循环策略

默认事件循环：

import asyncio

async def main():
    # Get running loop
    loop = asyncio.get_running_loop()
    print(f'Loop: {loop}')

asyncio.run(main())

自定义事件循环：

import asyncio

async def main():
    pass

# Create new event loop
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

try:
    loop.run_until_complete(main())
finally:
    loop.close()

事件循环最佳实践：

1. 简单程序使用 asyncio.run()（Python 3.7+）
2. 避免在事件循环之外创建 ClientSession
3. 使用完毕后务必关闭事件循环
4. 不要在事件循环中调用阻塞函数

运行阻塞代码

使用 ThreadPoolExecutor：

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor

def blocking_io():
    # Blocking operation
    time.sleep(2)
    return 'Done'

async def main():
    loop = asyncio.get_running_loop()

    # Run blocking code in thread pool
    result = await loop.run_in_executor(
        None,  # Use default executor
        blocking_io
    )

    return result

asyncio.run(main())

自定义执行器：

import asyncio
from concurrent.futures import ThreadPoolExecutor

async def main():
    loop = asyncio.get_running_loop()

    # Custom executor with 4 threads
    with ThreadPoolExecutor(max_workers=4) as executor:
        results = await asyncio.gather(*[
            loop.run_in_executor(executor, blocking_io)
            for _ in range(10)
        ])

    return results

asyncio.run(main())

循环回调

安排回调：

import asyncio

def callback(arg):
    print(f'Callback called with {arg}')

async def main():
    loop = asyncio.get_running_loop()

    # Schedule callback
    loop.call_soon(callback, 'immediate')

    # Schedule with delay
    loop.call_later(2, callback, 'delayed')

    # Schedule at specific time
    loop.call_at(loop.time() + 3, callback, 'scheduled')

    await asyncio.sleep(4)

asyncio.run(main())

异步上下文管理器

创建异步上下文管理器

基于类：

import asyncio

class AsyncDatabaseConnection:
    def __init__(self, host):
        self.host = host
        self.connection = None

    async def __aenter__(self):
        print(f'Connecting to {self.host}')
        await asyncio.sleep(0.1)  # Simulate connection
        self.connection = f'Connection to {self.host}'
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print(f'Closing connection to {self.host}')
        await asyncio.sleep(0.1)  # Simulate cleanup
        self.connection = None

    async def query(self, sql):
        if not self.connection:
            raise RuntimeError('Not connected')
        await asyncio.sleep(0.05)
        return f'Results for: {sql}'

async def main():
    async with AsyncDatabaseConnection('localhost') as db:
        result = await db.query('SELECT * FROM users')
        print(result)

asyncio.run(main())

基于装饰器：

import asyncio
from contextlib import asynccontextmanager

@asynccontextmanager
async def async_resource(name):
    # Setup
    print(f'Acquiring {name}')
    await asyncio.sleep(0.1)

    try:
        yield name
    finally:
        # Cleanup
        print(f'Releasing {name}')
        await asyncio.sleep(0.1)

async def main():
    async with async_resource('database') as db:
        print(f'Using {db}')

asyncio.run(main())

真实案例：aiohttp ClientSession

import aiohttp
import asyncio

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    # ClientSession as async context manager
    async with aiohttp.ClientSession() as session:
        html = await fetch(session, 'http://python.org')
        print(f'Body: {html[:100]}...')

asyncio.run(main())

为什么要为 ClientSession 使用异步上下文管理器？

1. 确保正确清理连接
2. 防止资源泄漏
3. 正确管理 SSL 连接
4. 处理优雅关闭

性能优化

分析异步代码

基本计时：

import asyncio
import time

async def slow_operation():
    await asyncio.sleep(1)

async def main():
    start = time.perf_counter()

    await slow_operation()

    elapsed = time.perf_counter() - start
    print(f'Took {elapsed:.2f} seconds')

asyncio.run(main())

分析多个操作：

import asyncio
import time

async def timed_task(name, duration):
    start = time.perf_counter()
    await asyncio.sleep(duration)
    elapsed = time.perf_counter() - start
    print(f'{name} took {elapsed:.2f}s')
    return name

async def main():
    await asyncio.gather(
        timed_task('Task 1', 1),
        timed_task('Task 2', 2),
        timed_task('Task 3', 0.5)
    )

asyncio.run(main())

优化并发

不佳——顺序执行：

async def slow_approach():
    results = []
    for i in range(10):
        result = await fetch_data(i)
        results.append(result)
    return results
# Takes 10 * fetch_time

良好——并发执行：

async def fast_approach():
    tasks = [fetch_data(i) for i in range(10)]
    results = await asyncio.gather(*tasks)
    return results
# Takes ~fetch_time

更佳——受控并发：

async def controlled_approach():
    semaphore = asyncio.Semaphore(5)  # Max 5 concurrent

    async def fetch_with_limit(i):
        async with semaphore:
            return await fetch_data(i)

    tasks = [fetch_with_limit(i) for i in range(10)]
    results = await asyncio.gather(*tasks)
    return results
# Takes ~2 * fetch_time, but respects limits

避免常见的性能问题

1. 不要为每个请求创建会话：

# BAD - Creates new session each time
async def bad_fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            return await resp.text()

# GOOD - Reuse session
async def good_fetch():
    async with aiohttp.ClientSession() as session:
        results = await asyncio.gather(
            session.get('http://example.com/1'),
            session.get('http://example.com/2'),
            session.get('http://example.com/3')
        )
        return results

2. 不要使用阻塞操作：

import asyncio
import requests  # Blocking library

# BAD - Blocks event loop
async def bad_request():
    response = requests.get('http://example.com')  # BLOCKS!
    return response.text

# GOOD - Use async library
async def good_request():
    async with aiohttp.ClientSession() as session:
        async with session.get('http://example.com') as resp:
            return await resp.text()

# ACCEPTABLE - If must use blocking, use executor
async def acceptable_request():
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(
        None,
        lambda: requests.get('http://example.com').text
    )
    return result

3. 使用零睡眠正确清理：

async def proper_cleanup():
    async with aiohttp.ClientSession() as session:
        async with session.get('http://example.org/') as resp:
            await resp.read()

    # Zero-sleep to allow underlying connections to close
    await asyncio.sleep(0)

常见问题

问题 1：在事件循环之外创建 ClientSession

问题：

import aiohttp

# BAD - Session created outside event loop
session = aiohttp.ClientSession()

async def fetch(url):
    async with session.get(url) as resp:
        return await resp.text()

为什么不好：

• 会话在创建时会绑定到事件循环
• 如果事件循环发生变化（例如，使用 uvloop），会话就会失效
• 可能导致程序挂起

解决方案：

import aiohttp
import asyncio

async def main():
    # Create session inside async function
    async with aiohttp.ClientSession() as session:
        async with session.get('http://python.org') as resp:
            print(await resp.text())

asyncio.run(main())

问题 2：将会话用作类变量

问题：

class API:
    session = aiohttp.ClientSession()  # BAD - global instance

    async def fetch(self, url):
        async with self.session.get(url) as resp:
            return await resp.text()

解决方案：

class API:
    def __init__(self):
        self.session = None

    async def __aenter__(self):
        self.session = aiohttp.ClientSession()
        return self

    async def __aexit__(self, *args):
        await self.session.close()

    async def fetch(self, url):
        async with self.session.get(url) as resp:
            return await resp.text()

# Usage
async def main():
    async with API() as api:
        result = await api.fetch('http://example.com')

问题 3：忘记使用 await

问题：

async def process_data():
    # Forgot await - returns coroutine, doesn't execute!
    result = fetch_data()  # Missing await
    return result

解决方案：

async def process_data():
    result = await fetch_data()  # Proper await
    return result

问题 4：阻塞事件循环

问题：

import asyncio
import time

async def bad_sleep():
    time.sleep(5)  # BAD - Blocks entire event loop!

async def main():
    await asyncio.gather(
        bad_sleep(),
        another_task()  # Blocked for 5 seconds
    )

解决方案：

import asyncio

async def good_sleep():
    await asyncio.sleep(5)  # GOOD - Yields control

async def main():
    await asyncio.gather(
        good_sleep(),
        another_task()  # Runs concurrently
    )

问题 5：未处理任务取消

问题：

async def bad_task():
    while True:
        await asyncio.sleep(1)
        process_data()
        # No cleanup on cancellation!

解决方案：

async def good_task():
    try:
        while True:
            await asyncio.sleep(1)
            process_data()
    except asyncio.CancelledError:
        # Cleanup resources
        cleanup()
        raise  # Re-raise to mark as cancelled

问题 6：锁导致死锁

问题：

import asyncio

lock1 = asyncio.Lock()
lock2 = asyncio.Lock()

async def task_a():
    async with lock1:
        await asyncio.sleep(0.1)
        async with lock2:  # Deadlock potential
            pass

async def task_b():
    async with lock2:
        await asyncio.sleep(0.1)
        async with lock1:  # Deadlock potential
            pass

解决方案：

# Always acquire locks in same order
async def safe_task_a():
    async with lock1:
        async with lock2:
            pass

async def safe_task_b():
    async with lock1:  # Same order
        async with lock2:
            pass

生产环境模式

模式 1：优雅关闭

完整关闭示例：

import asyncio
import signal
from contextlib import suppress

class Application:
    def __init__(self):
        self.should_exit = False
        self.tasks = []

    async def worker(self, name):
        try:
            while not self.should_exit:
                print(f'{name} working...')
                await asyncio.sleep(1)
        except asyncio.CancelledError:
            print(f'{name} cancelled, cleaning up...')
            raise

    def handle_signal(self, sig):
        print(f'Received signal {sig}, shutting down...')
        self.should_exit = True

    async def run(self):
        # Setup signal handlers
        loop = asyncio.get_running_loop()
        for sig in (signal.SIGTERM, signal.SIGINT):
            loop.add_signal_handler(
                sig,
                lambda s=sig: self.handle_signal(s)
            )

        # Start workers
        self.tasks = [
            asyncio.create_task(self.worker(f'Worker-{i}'))
            for i in range(3)
        ]

        # Wait for shutdown signal
        while not self.should_exit:
            await asyncio.sleep(0.1)

        # Cancel all tasks
        for task in self.tasks:
            task.cancel()

        # Wait for cancellation to complete
        await asyncio.gather(*self.tasks, return_exceptions=True)

        print('Shutdown complete')

# Run application
app = Application()
asyncio.run(app.run())

模式 2：结合应用生命周期的后台任务

包含后台任务的 aiohttp 应用：

import asyncio
from contextlib import suppress
from aiohttp import web

async def listen_to_redis(app):
    """Background task that listens to Redis"""
    # Simulated Redis listening
    try:
        while True:
            # Process messages
            await asyncio.sleep(1)
            print('Processing Redis message...')
    except asyncio.CancelledError:
        print('Redis listener stopped')
        raise

async def background_tasks(app):
    """Cleanup context for managing background tasks"""
    # Startup: Create background task
    app['redis_listener'] = asyncio.create_task(listen_to_redis(app))

    yield  # App is running

    # Cleanup: Cancel background task
    app['redis_listener'].cancel()
    with suppress(asyncio.CancelledError):
        await app['redis_listener']

# Setup application
app = web.Application()
app.cleanup_ctx.append(background_tasks)

模式 3：带指数退避的重试逻辑

import asyncio
import aiohttp
from typing import Any, Callable

async def retry_with_backoff(
    coro_func: Callable,
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    *args,
    **kwargs
) -> Any:
    """
    Retry async function with exponential backoff

    Args:
        coro_func: Async function to retry
        max_retries: Maximum number of retries
        base_delay: Initial delay between retries
        max_delay: Maximum delay between retries
    """
    for attempt in range(max_retries):
        try:
            return await coro_func(*args, **kwargs)
        except Exception as e:
            if attempt == max_retries - 1:
                # Last attempt failed
                raise

            # Calculate delay with exponential backoff
            delay = min(base_delay * (2 ** attempt), max_delay)

            print(f'Attempt {attempt + 1} failed: {e}')
            print(f'Retrying in {delay:.1f} seconds...')

            await asyncio.sleep(delay)

# Usage
async def unstable_api_call():
    async with aiohttp.ClientSession() as session:
        async with session.get('http://unstable-api.com') as resp:
            return await resp.json()

async def main():
    result = await retry_with_backoff(
        unstable_api_call,
        max_retries=5,
        base_delay=1.0
    )
    return result

模式 4：熔断器

import asyncio
from datetime import datetime, timedelta
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"      # Normal operation
    OPEN = "open"          # Failing, reject requests
    HALF_OPEN = "half_open"  # Testing if recovered

class CircuitBreaker:
    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: float = 60.0,
        success_threshold: int = 2
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.success_threshold = success_threshold

        self.failure_count = 0
        self.success_count = 0
        self.state = CircuitState.CLOSED
        self.opened_at = None

    async def call(self, coro_func, *args, **kwargs):
        if self.state == CircuitState.OPEN:
            # Check if should try recovery
            if datetime.now() - self.opened_at > timedelta(seconds=self.recovery_timeout):
                self.state = CircuitState.HALF_OPEN
                self.success_count = 0
            else:
                raise Exception('Circuit breaker is OPEN')

        try:
            result = await coro_func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise

    def _on_success(self):
        self.failure_count = 0

        if self.state == CircuitState.HALF_OPEN:
            self.success_count += 1
            if self.success_count >= self.success_threshold:
                self.state = CircuitState.CLOSED
                self.success_count = 0

    def _on_failure(self):
        self.failure_count += 1

        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN
            self.opened_at = datetime.now()

# Usage
async def flaky_service():
    # Simulated flaky service
    import random
    await asyncio.sleep(0.1)
    if random.random() < 0.5:
        raise Exception('Service error')
    return 'Success'

async def main():
    breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=5.0)

    for i in range(20):
        try:
            result = await breaker.call(flaky_service)
            print(f'Request {i}: {result} - State: {breaker.state.value}')
        except Exception as e:
            print(f'Request {i}: Failed - State: {breaker.state.value}')

        await asyncio.sleep(0.5)

模式 5：包含多个事件源的 WebSocket

处理并行的 WebSocket 和后台事件：

import asyncio
from aiohttp import web

async def read_subscription(ws, redis):
    """Background task reading from Redis and sending to WebSocket"""
    # Simulated Redis subscription
    channel = await redis.subscribe('channel:1')

    try:
        # Simulate receiving messages
        for i in range(10):
            await asyncio.sleep(1)
            message = f'Redis message {i}'
            await ws.send_str(message)
    finally:
        await redis.unsubscribe('channel:1')

async def websocket_handler(request):
    """WebSocket handler with parallel event sources"""
    ws = web.WebSocketResponse()
    await ws.prepare(request)

    # Create background task for Redis subscription
    redis = request.app['redis']
    task = asyncio.create_task(read_subscription(ws, redis))

    try:
        # Handle incoming WebSocket messages
        async for msg in ws:
            if msg.type == web.WSMsgType.TEXT:
                # Process incoming message
                await ws.send_str(f'Echo: {msg.data}')
            elif msg.type == web.WSMsgType.ERROR:
                print(f'WebSocket error: {ws.exception()}')
    finally:
        # Cleanup: Cancel background task
        task.cancel()

    return ws

最佳实践

测试异步代码

使用 pytest-asyncio：

import pytest
import asyncio

@pytest.mark.asyncio
async def test_async_function():
    result = await async_operation()
    assert result == 'expected'

@pytest.mark.asyncio
async def test_with_fixture(aiohttp_client):
    client = await aiohttp_client(create_app())
    resp = await client.get('/')
    assert resp.status == 200

手动设置事件循环：

import asyncio
import unittest

class TestAsyncCode(unittest.TestCase):
    def setUp(self):
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self.loop)

    def tearDown(self):
        self.loop.close()

    def test_coroutine(self):
        async def test_impl():
            result = await async_function()
            self.assertEqual(result, 'expected')

        self.loop.run_until_complete(test_impl())

调试异步代码

启用调试模式：

import asyncio
import warnings

# Enable asyncio debug mode
asyncio.run(main(), debug=True)

# Or manually
loop = asyncio.get_event_loop()
loop.set_debug(True)
loop.run_until_complete(main())

调试模式可检测：

• 从未被 await 的协程
• 执行耗时过长的回调
• 仍处于等待状态时被销毁的任务

记录运行缓慢的回调：

import asyncio
import logging

logging.basicConfig(level=logging.DEBUG)

loop = asyncio.get_event_loop()
loop.slow_callback_duration = 0.1  # 100ms threshold
loop.set_debug(True)

文档

为异步函数编写文档：

async def fetch_user_data(user_id: int) -> dict:
    """
    Fetch user data from the database.

    Args:
        user_id: The unique identifier of the user

    Returns:
        Dictionary containing user data

    Raises:
        UserNotFoundError: If user doesn't exist
        DatabaseError: If database connection fails

    Example:
        >>> async def main():
        ...     user = await fetch_user_data(123)
        ...     print(user['name'])

    Note:
        This function must be called within an async context.
        Connection pooling is handled automatically.
    """
    async with get_db_connection() as conn:
        return await conn.fetch_one(
            'SELECT * FROM users WHERE id = $1',
            user_id
        )

完整示例

示例 1：并行 HTTP 请求

import asyncio
import aiohttp
import time

async def fetch(session, url):
    """Fetch a single URL"""
    async with session.get(url) as response:
        return {
            'url': url,
            'status': response.status,
            'length': len(await response.text())
        }

async def fetch_all(urls):
    """Fetch multiple URLs concurrently"""
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        return results

async def main():
    urls = [
        'http://python.org',
        'http://docs.python.org',
        'http://pypi.org',
        'http://github.com/python',
        'http://www.python.org/dev/peps/'
    ]

    start = time.perf_counter()
    results = await fetch_all(urls)
    elapsed = time.perf_counter() - start

    for result in results:
        print(f"{result['url']}: {result['status']} ({result['length']} bytes)")

    print(f"\nFetched {len(urls)} URLs in {elapsed:.2f} seconds")

asyncio.run(main())

示例 2：限速 API 客户端

import asyncio
import aiohttp
from typing import List, Dict, Any

class RateLimitedClient:
    def __init__(self, rate_limit: int = 10):
        """
        Args:
            rate_limit: Maximum concurrent requests
        """
        self.semaphore = asyncio.Semaphore(rate_limit)
        self.session = None

    async def __aenter__(self):
        self.session = aiohttp.ClientSession()
        return self

    async def __aexit__(self, *args):
        await self.session.close()
        # Allow connections to close
        await asyncio.sleep(0)

    async def fetch(self, url: str) -> Dict[str, Any]:
        """Fetch URL with rate limiting"""
        async with self.semaphore:
            print(f'Fetching {url}')
            async with self.session.get(url) as resp:
                return {
                    'url': url,
                    'status': resp.status,
                    'data': await resp.json()
                }

    async def fetch_all(self, urls: List[str]) -> List[Dict[str, Any]]:
        """Fetch all URLs with rate limiting"""
        tasks = [self.fetch(url) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

async def main():
    urls = [f'https://api.github.com/users/{user}'
            for user in ['python', 'django', 'flask', 'requests', 'aiohttp']]

    async with RateLimitedClient(rate_limit=2) as client:
        results = await client.fetch_all(urls)

        for result in results:
            if isinstance(result, Exception):
                print(f'Error: {result}')
            else:
                print(f"User: {result['data'].get('login', 'unknown')}")

asyncio.run(main())

示例 3：数据库连接池

import asyncio
from typing import List, Any

class AsyncConnectionPool:
    def __init__(self, size: int = 10):
        self.pool = asyncio.Queue(maxsize=size)
        self.size = size

    async def init(self):
        """Initialize connection pool"""
        for i in range(self.size):
            conn = await self._create_connection(i)
            await self.pool.put(conn)

    async def _create_connection(self, conn_id: int):
        """Create a database connection (simulated)"""
        await asyncio.sleep(0.1)  # Simulate connection time
        return {'id': conn_id, 'connected': True}

    async def acquire(self):
        """Acquire connection from pool"""
        return await self.pool.get()

    async def release(self, conn):
        """Release connection back to pool"""
        await self.pool.put(conn)

    async def execute(self, query: str) -> Any:
        """Execute query using pooled connection"""
        conn = await self.acquire()
        try:
            # Simulate query execution
            await asyncio.sleep(0.05)
            return f"Query '{query}' executed on connection {conn['id']}"
        finally:
            await self.release(conn)

    async def close(self):
        """Close all connections"""
        while not self.pool.empty():
            conn = await self.pool.get()
            # Close connection (simulated)
            conn['connected'] = False

async def worker(pool: AsyncConnectionPool, worker_id: int):
    """Worker that executes queries"""
    for i in range(5):
        result = await pool.execute(f'SELECT * FROM table WHERE id={i}')
        print(f'Worker {worker_id}: {result}')

async def main():
    # Create and initialize pool
    pool = AsyncConnectionPool(size=5)
    await pool.init()

    # Run multiple workers concurrently
    await asyncio.gather(*[
        worker(pool, i) for i in range(10)
    ])

    # Cleanup
    await pool.close()

asyncio.run(main())

示例 4：实时数据处理器

import asyncio
import random
from datetime import datetime

class DataProcessor:
    def __init__(self):
        self.queue = asyncio.Queue()
        self.processed = 0
        self.errors = 0

    async def producer(self, producer_id: int):
        """Produce data items"""
        for i in range(10):
            await asyncio.sleep(random.uniform(0.1, 0.5))
            item = {
                'producer_id': producer_id,
                'item_id': i,
                'timestamp': datetime.now(),
                'data': random.randint(1, 100)
            }
            await self.queue.put(item)
            print(f'Producer {producer_id} generated item {i}')

        # Signal completion
        await self.queue.put(None)

    async def consumer(self, consumer_id: int):
        """Consume and process data items"""
        while True:
            item = await self.queue.get()

            if item is None:
                # Propagate sentinel
                await self.queue.put(None)
                break

            try:
                # Simulate processing
                await asyncio.sleep(random.uniform(0.05, 0.2))

                # Process item
                result = item['data'] * 2
                print(f"Consumer {consumer_id} processed: {item['item_id']} -> {result}")

                self.processed += 1
            except Exception as e:
                print(f'Consumer {consumer_id} error: {e}')
                self.errors += 1
            finally:
                self.queue.task_done()

    async def monitor(self):
        """Monitor processing statistics"""
        while True:
            await asyncio.sleep(2)
            print(f'\n=== Stats: Processed={self.processed}, Errors={self.errors}, Queue={self.queue.qsize()} ===\n')

    async def run(self, num_producers: int = 3, num_consumers: int = 5):
        """Run the data processor"""
        # Start monitor
        monitor_task = asyncio.create_task(self.monitor())

        # Start producers and consumers
        await asyncio.gather(
            *[self.producer(i) for i in range(num_producers)],
            *[self.consumer(i) for i in range(num_consumers)]
        )

        # Cancel monitor
        monitor_task.cancel()

        print(f'\nFinal Stats: Processed={self.processed}, Errors={self.errors}')

async def main():
    processor = DataProcessor()
    await processor.run(num_producers=3, num_consumers=5)

asyncio.run(main())

示例 5：使用 aiofiles 进行异步文件 I/O

import asyncio
import aiofiles
from pathlib import Path

async def write_file(path: str, content: str):
    """Write content to file asynchronously"""
    async with aiofiles.open(path, 'w') as f:
        await f.write(content)

async def read_file(path: str) -> str:
    """Read file content asynchronously"""
    async with aiofiles.open(path, 'r') as f:
        return await f.read()

async def process_files(file_paths: list):
    """Process multiple files concurrently"""
    tasks = [read_file(path) for path in file_paths]
    contents = await asyncio.gather(*tasks)

    # Process contents
    results = []
    for path, content in zip(file_paths, contents):
        result = {
            'path': path,
            'lines': len(content.split('\n')),
            'words': len(content.split()),
            'chars': len(content)
        }
        results.append(result)

    return results

async def main():
    # Create test files
    test_files = ['test1.txt', 'test2.txt', 'test3.txt']

    # Write files concurrently
    await asyncio.gather(*[
        write_file(f, f'Content of file {f}\n' * 10)
        for f in test_files
    ])

    # Process files
    results = await process_files(test_files)

    for result in results:
        print(f"{result['path']}: {result['lines']} lines, "
              f"{result['words']} words, {result['chars']} chars")

    # Cleanup
    for f in test_files:
        Path(f).unlink(missing_ok=True)

# asyncio.run(main())  # Uncomment to run (requires aiofiles)

资源

• Python asyncio 文档：https://docs.python.org/3/library/asyncio.html
• aiohttp 文档：https://docs.aiohttp.org/
• Real Python asyncio 指南：https://realpython.com/async-io-python/
• PEP 492 - 使用 async 和 await 语法的协程：https://www.python.org/dev/peps/pep-0492/
• asyncio 速查表：https://www.pythonsheets.com/notes/python-asyncio.html
• Effective Python：条目 60 - 考虑使用 asyncio：https://effectivepython.com/

---

技能版本：1.0.0 最后更新：2025 年 10 月 技能类别：并发、性能、异步编程 兼容对象：Python 3.7+、aiohttp、asyncio、uvloop