异步编程

What — 是什么

Python 的异步编程基于 asyncio 库，使用 async/await 语法编写协程，由事件循环调度执行，实现单线程并发 I/O。

核心概念：

• 协程（Coroutine）：async def 定义的函数，调用后返回协程对象
• 事件循环（Event Loop）：asyncio.run() 创建，调度协程执行
• Task：对协程的封装，可并发调度和取消
• Future：表示异步操作最终结果的占位对象

关键特性：

• await 挂起当前协程，让出控制权给事件循环
• asyncio.gather() 并发执行多个协程
• asyncio.create_task() 将协程包装为 Task 立即调度

运行机制：

• 内存模型：协程在用户态切换，无线程栈开销
• 执行模型：单线程事件循环，遇到 I/O 就切走，I/O 完成后恢复
• 并发模型：协程（协作式调度），非抢占式

类型系统：

• 类型分类：Coroutine、AsyncGenerator、Awaitable
• 类型转换规则：async def 函数返回 Coroutine[Any, Any, T]
• 泛型/多态支持：可用 Awaitable[T] 作为通用异步类型

Why — 为什么

适用场景：

• 高并发网络请求（爬虫、API 调用）
• 数据库/Redis 异步驱动
• WebSocket 服务
• 异步文件 I/O（aiofiles）

对比其他语言：

优缺点：

• ✅ 优点：
  • 单线程避免锁和竞争
  • 代码结构清晰（async/await）
  • 与类型提示配合良好
• ❌ 缺点：
  • 生态不如同步库丰富（部分库无异步版本）
  • GIL 限制 CPU 并行，需用多进程补充
  • 同步/异步代码不能混用，否则阻塞事件循环

How — 怎么用

快速上手

import asyncio

async def fetch_data(url: str) -> str:
    await asyncio.sleep(1)  # 模拟 I/O
    return f"data from {url}"

async def main():
    # 并发请求
    results = await asyncio.gather(
        fetch_data("api/a"),
        fetch_data("api/b"),
        fetch_data("api/c"),
    )
    print(results)

asyncio.run(main())

代码示例

超时与取消：

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

async def main():
    try:
        await asyncio.wait_for(slow_operation(), timeout=3.0)
    except asyncio.TimeoutError:
        print("操作超时")

生产者-消费者：

async def producer(queue: asyncio.Queue):
    for i in range(5):
        await asyncio.sleep(0.1)
        await queue.put(f"item-{i}")

async def consumer(queue: asyncio.Queue, name: str):
    while True:
        item = await queue.get()
        print(f"{name} 处理: {item}")
        queue.task_done()

async def main():
    queue = asyncio.Queue(maxsize=3)
    await asyncio.gather(
        producer(queue),
        consumer(queue, "C1"),
        consumer(queue, "C2"),
    )

常见问题与踩坑

最佳实践

• I/O 密集用 async，CPU 密集用多进程
• 使用 asyncio.gather 替代顺序 await
• 第三方库优先选择异步版本
• 用 asyncio.to_thread() 在协程中调用同步阻塞函数

面试题

Q1: async/await 的原理是什么？

async def 定义协程函数，调用后返回协程对象（不会立即执行）；await 挂起当前协程并将控制权交还给事件循环，等待可等待对象（协程/Task/Future）完成后恢复执行。本质是协程间的协作式调度，而非操作系统级别的线程切换。

Q2: 事件循环的工作机制是什么？

事件循环是 asyncio 的核心调度器，运行在单线程中。它不断轮询就绪的 I/O 事件和定时器，将就绪的 Task 恢复执行；当协程遇到 await 时挂起并注册回调，事件循环转而调度其他就绪的 Task，实现 I/O 并发。

Q3: 什么时候用 asyncio，什么时候用多线程？

I/O 密集型任务优先用 asyncio（网络请求、数据库查询），协程切换开销极小且无线程安全问题；需要与现有同步库交互或使用阻塞 I/O 时用多线程；CPU 密集型任务两者都不适合，应使用多进程（ProcessPoolExecutor）。

Q4: 协程和线程的区别是什么？

协程是用户态的协作式调度，由开发者通过 await 主动让出控制权，切换开销极小，无线程安全问题；线程是操作系统调度的抢占式并发，切换开销较大，需要加锁保护共享资源。协程在单线程中运行，无法利用多核 CPU。

Q5: 在异步代码中调用同步阻塞函数会怎样？如何解决？

同步阻塞函数（如 time.sleep()、requests.get()）会阻塞整个事件循环，导致所有协程都无法执行。解决方案：使用 asyncio.to_thread() 将同步函数放到线程池执行，或使用异步替代库（aiohttp 替代 requests）。

Q6: asyncio.gather 和 asyncio.TaskGroup 有什么区别？

asyncio.gather() 是传统方式，按顺序收集结果，异常需手动处理（return_exceptions=True）。asyncio.TaskGroup（Python 3.11+）是上下文管理器，任何任务抛异常时自动取消其他任务，保证所有任务完成后才退出，异常作为 ExceptionGroup 抛出。TaskGroup 是推荐的新方式——更安全（自动取消）、更清晰（结构化并发）。

Q7: asyncio.Queue 和多线程 Queue 有什么区别？

asyncio.Queue 是协程安全的，用 await queue.get() 和 await queue.put() 异步等待，不阻塞事件循环。queue.Queue 是线程安全的，用 queue.get() 同步等待，会阻塞当前线程。混用会导致事件循环阻塞。异步场景必须用 asyncio.Queue，多线程场景用 queue.Queue。

Q8: 什么是结构化并发（Structured Concurrency）？Python 如何实现？

结构化并发要求所有并发任务有明确的生命周期——任务在某个作用域内创建，必须在作用域结束前完成或取消，不能”泄漏”到外部。Python 3.11 的 asyncio.TaskGroup 实现了结构化并发：进入 with 块时创建任务，退出时等待全部完成（任一失败则取消其余）。好处：无孤儿任务、异常不丢失、资源不泄漏。

相关链接：

• 装饰器与闭包
• GIL与并发模型
• FastAPI
• WebSocket与实时通信
• Python asyncio 文档