---
id: python.asyncio.interview
title: asyncio 面试题
type: interview
domain: python
stack: [asyncio]
tags: [asyncio, 协程, 并发, 面试]
level: intermediate
related: [python.asyncio.coroutines-tasks, python.asyncio.concurrency-control]
week: [4, 16]
---

# asyncio 面试题

## 基础

### 【基础】协程、线程、进程有什么区别？

**参考答案**：进程有独立的内存空间，切换和通信开销大，能利用多核；线程共享进程内存，由操作系统抢占式调度，在 CPython 中受 GIL 限制不能并行执行 Python 代码；协程是用户态的、协作式调度的执行单元，在 `await` 处主动让出控制权，切换开销极小，单线程即可同时处理大量 IO 等待。

**追问**：Python 多线程适合什么场景？→ I/O 密集且必须使用同步库的场景；CPU 密集任务因为 GIL 用多进程。

### 【基础】asyncio 适合什么场景？不适合什么场景？

**参考答案**：适合 IO 密集、并发连接多的场景：调用大模型 API、HTTP 请求、数据库查询、WebSocket / SSE 长连接。不适合 CPU 密集计算（单线程，无法加速），这类任务用多进程或交给 C 扩展库。

**追问**：CPU 密集的任务混在异步服务里怎么办？→ 用 `loop.run_in_executor` 放到进程池，或者交给单独的任务队列 / 服务处理，避免阻塞事件循环。

### 【基础】调用 `async def` 函数但没有 `await` 会怎样？

**参考答案**：只会创建协程对象，函数体不会执行，Python 会给出 “coroutine was never awaited” 警告。这一点和 JS 不同——JS 的 async 函数调用后立即开始执行。

**追问**：怎么发现忘记 await 的问题？→ 运行时会出现 `coroutine was never awaited` 的警告；类型检查工具（mypy / pyright）和 linter 也能检查出来。

### 【基础】为什么在协程里不能使用 `time.sleep` 或 `requests`？

**参考答案**：事件循环是单线程的，阻塞调用会让整个线程停住，所有其他协程都无法运行，服务的并发能力退化为 1。应该用 `asyncio.sleep`、`httpx.AsyncClient` 等异步库；只能用同步库时用 `asyncio.to_thread` 放到线程池执行。

**追问**：必须使用同步库怎么办？→ 用 `asyncio.to_thread()` 放到线程池执行，或者换成异步库（`httpx.AsyncClient`、`asyncpg` 等）。

## 进阶

### 【进阶】`asyncio.gather` 和 `TaskGroup` 有什么区别？

**参考答案**：`gather` 在某个任务抛异常时立即把异常抛给调用方，但其余任务继续运行；加 `return_exceptions=True` 则把异常作为结果返回。`TaskGroup`（3.11+）是结构化并发：一个任务失败时取消组内其他任务，等它们全部结束后，以 `ExceptionGroup` 抛出所有异常，不会留下“孤儿任务”。

**追问**：`gather` 的 `return_exceptions=True` 有什么用？→ 一个任务异常时不会立即抛出，而是把异常作为结果返回，适合“部分失败可以接受”的批量调用。

### 【进阶】怎么限制并发数？

**参考答案**：用 `asyncio.Semaphore(n)`，在每个任务的实际调用外层 `async with sem:`；或者用固定数量的 worker 从 `asyncio.Queue` 取任务。注意多个请求要共享同一个 Semaphore；多进程 / 多实例时需要 Redis 等全局限流。

**追问**：信号量和限速有什么区别？→ 信号量限制同时进行的请求数；限速限制单位时间内的请求数（如每分钟 600 次），服务商通常两种都有限制，需要同时控制。

### 【进阶】asyncio 的取消机制是怎样的？

**参考答案**：`task.cancel()` 请求取消，任务在下一个 `await` 点收到 `CancelledError`。任务可以捕获它做清理（关闭连接、记录状态），但应该重新抛出，否则取消失效。`asyncio.timeout` / `wait_for` 超时也是通过取消内部任务实现的。`CancelledError` 继承自 `BaseException`，所以 `except Exception` 不会误吞它。

**追问**：捕获了 `CancelledError` 后应该怎么做？→ 做完清理后重新抛出；吞掉它会导致取消失效、`TaskGroup` / 超时无法正常工作。

## 场景题

### 【场景题】需要对 1 万条文本调用大模型生成摘要，服务商限制每分钟 600 次请求，怎么设计？

**参考答案**：用 Semaphore 或 worker 池限制并发，再加速率控制（令牌桶，每秒约 10 次）；每次调用设超时；对 429、超时、5xx 做带抖动的指数退避重试，429 响应如果带 `Retry-After` 就按它等待；单条失败记录下来，最后汇总重跑；记录进度（已完成的 ID）以便中断后续跑；结果写入数据库而不是全放内存。规模更大时用消息队列分发给多个 worker。

**追问**：中途失败的任务怎么办？→ 每条记录保存处理状态，失败的记录按错误类型重试（限流错误退避重试，内容错误记录跳过），支持断点续跑。

### 【场景题】FastAPI 接口里调用了一个同步的 SDK，发现服务在压测时吞吐很低，为什么？怎么改？

**参考答案**：在 `async def` 路由里直接调用同步 SDK，阻塞了事件循环，所有请求排队。改法：换成 SDK 的异步版本；或者 `await asyncio.to_thread(sdk.call, ...)`；或者把路由定义为普通 `def`，FastAPI 会把它放到线程池执行。然后重新压测确认。

**追问**：`def` 路由和 `async def` 路由在这里有什么区别？→ `def` 路由会被 FastAPI 放到线程池执行，不会阻塞事件循环；`async def` 中直接调用同步 SDK 会阻塞整个事件循环。

### 【场景题】一个异步任务在服务关闭时总是卡住，可能是什么原因？

**参考答案**：任务里捕获了 `CancelledError` 却没有重新抛出（或者用了裸 `except:`），导致取消无效；或者任务在执行不响应取消的阻塞调用（同步 IO、`to_thread` 中的线程无法被强制停止）；或者有后台任务没有被追踪和取消。修复：正确传播取消、给阻塞调用设超时、在应用的生命周期关闭阶段统一取消并等待后台任务。

**追问**：怎么给关闭过程设置上限？→ 用 `asyncio.timeout()` 或 `asyncio.wait_for()` 给关闭等待设置超时，超时后取消剩余任务并记录日志。