3. Python 异步编程基础：async/await 语法、事件循环、协程与任务

在构建高吞吐的 Redis 缓存中间件时，I/O 等待（网络延迟、磁盘刷写）是性能瓶颈的核心来源。Python 的 async/await 语法提供了一种协作式多任务模型，让单线程在等待 I/O 时主动让出控制权，从而并发处理成百上千个连接。本章将拆解异步编程的四大基石：协程、任务、事件循环与 async/await 语法，并展示它们如何协同工作。

3.1 核心概念与关系

异步编程的四个核心元素构成一个分层协作体系。下表梳理了每个角色的职责与生命周期：

概念	角色	生命周期	类比（餐厅）
协程（Coroutine）	可暂停/恢复的函数	从 async def 调用到 return 或 StopIteration	菜谱步骤
任务（Task）	协程的并发调度单元	从 asyncio.create_task() 到协程完成或被取消	厨师手上的订单
事件循环（Event Loop）	全局调度器，管理所有任务	从 asyncio.run() 启动到所有任务完成	餐厅经理
await 表达式	挂起点，让出控制权	在协程内部，遇到 I/O 时触发	“等菜”信号

下图展示了事件循环如何驱动多个协程任务交错执行：

3.2 async/await 语法详解

async def 定义一个协程函数，调用它返回一个协程对象。await 只能在协程内部使用，用于挂起当前协程直到可等待对象（另一个协程、Future 或 Task）完成。

3.2.1 基础用法

import asyncio

async def fetch_redis(key: str) -> str:
    """模拟 Redis GET 操作"""
    await asyncio.sleep(0.1)  # 模拟网络 I/O
    return f"value_{key}"

async def main():
    # 直接 await 协程：顺序执行
    result1 = await fetch_redis("user:1")
    result2 = await fetch_redis("user:2")
    print(result1, result2)

asyncio.run(main())

3.2.2 并发执行：创建任务

使用 asyncio.create_task() 将协程包装为 Task，事件循环会在后台调度它，无需等待即可继续执行后续代码。

async def main():
    # 创建两个任务，并发执行
    task1 = asyncio.create_task(fetch_redis("user:1"))
    task2 = asyncio.create_task(fetch_redis("user:2"))

    # 等待所有任务完成
    result1 = await task1
    result2 = await task2
    print(result1, result2)

asyncio.run(main())

对比两种模式：

模式	执行方式	总耗时（假设每个 I/O 100ms）	适用场景
await 顺序执行	串行	200ms	依赖前一个结果
create_task + await	并发	~100ms	独立 I/O 请求

3.3 事件循环的启动与生命周期

asyncio.run() 是 Python 3.7+ 推荐的入口函数，它自动完成三件事：

1. 创建一个新的事件循环实例
2. 运行传入的协程直到完成
3. 关闭事件循环并清理资源

在更复杂的场景（如集成 Redis 客户端时），你可能需要手动管理事件循环：

async def redis_pipeline():
    # 模拟多个 Redis 命令
    tasks = [
        asyncio.create_task(fetch_redis("key1")),
        asyncio.create_task(fetch_redis("key2")),
        asyncio.create_task(fetch_redis("key3")),
    ]
    results = await asyncio.gather(*tasks)
    return results

# 手动管理事件循环（高级用法）
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
    results = loop.run_until_complete(redis_pipeline())
    print(results)
finally:
    loop.close()

3.4 协程与任务的深入对比

理解协程与任务的区别，是正确编排异步流程的关键：

特性	协程（Coroutine）	任务（Task）
本质	可暂停的函数对象	协程 + 调度上下文
创建方式	async def 调用	asyncio.create_task(coro)
是否自动调度	否，需要 await 或包装为 Task	是，创建后立即加入事件循环
取消支持	不支持直接取消	支持 task.cancel()
结果获取	await coro 直接返回	await task 或 task.result()
调试信息	有限	包含名称、状态、堆栈

下图展示了从协程定义到任务调度的完整生命周期：

3.5 异步上下文管理器与异步迭代器

在 Redis 客户端封装中，连接池的获取与释放非常适合使用异步上下文管理器：

class RedisConnection:
    async def __aenter__(self):
        self.conn = await create_redis_connection()
        return self.conn

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.conn.close()

# 使用
async def main():
    async with RedisConnection() as conn:
        value = await conn.get("key")
        print(value)

异步迭代器则适用于处理 Redis 的 SCAN 或流式数据：

class RedisScanner:
    def __init__(self, pattern: str):
        self.pattern = pattern
        self.cursor = 0

    def __aiter__(self):
        return self

    async def __anext__(self):
        # 模拟 SCAN 迭代
        if self.cursor > 100:
            raise StopAsyncIteration
        self.cursor += 10
        return f"key_{self.cursor}"

async def main():
    async for key in RedisScanner("user:*"):
        print(f"Found: {key}")

3.6 常见陷阱与调试技巧

• 忘记 await：协程对象未被调度，日志中会出现 coroutine was never awaited 警告。
• 阻塞事件循环：在协程内使用 time.sleep() 会阻塞整个事件循环，应使用 asyncio.sleep()。
• 任务未捕获异常：Task 内部的异常不会自动传播，必须 await task 或调用 task.exception() 才能捕获。
• 调试工具：设置 PYTHONASYNCIODEBUG=1 环境变量，或调用 loop.set_debug(True) 可输出详细的调度日志。