去年我还是一个习惯写同步代码的人，总以为requests.get()卡一下很正常、接口响应慢就多设超时时间。直到某天需要对上千个内部服务做健康检查，跑一次要十几分钟，这还是在机器空闲的情况下。把业务切到asyncio之后，同样的任务压缩到十秒量级，那一刻我才真正意识到：Python 异步编程不是简单的 API 切换，而是对“等待”这件事的重构。今天围绕 Python 异步编程中最重要的 Asyncio 库，把这个过程里该理解的概念、该写的代码、该避开的坑，从头到尾翻一遍。

这篇内容不是从零讲语法，而是按照“理解问题 → 理清核心概念 → 写出可运行代码 → 落到真实场景 → 处理混合负载 → 解决疑难杂症”的顺序展开。适合两类人：一类是刚接触异步、想系统性入门的 Python 开发者；另一类是已经在写异步但总被莫名其妙的问题卡住的同学，分享里很多问题是我实际踩过并排查过的，希望能替你省下一点时间。

1. 先弄明白：异步到底在解决什么问题

1.1 同步代码为什么慢：都在等

先看一个最简单的场景。写一个函数，从三个 HTTP API 拉数据，每个接口耗时 1 秒，最后一个循环里依次await三个调用。这段代码本身不难，难的是理解它“为什么慢”。

同步版本长这样：代码发起网络请求之后，整个线程就挂在原地等网络返回。在这个等待的过程里，CPU 其实是完全空闲的，内存也没多大压力，但整个程序却像被按了暂停键。如果你同时发起 100 个请求，同步代码就是一个一个等，总耗时接近所有请求耗时之和。

异步要做的事情很简单：把“等待不做事”的时间利用起来。当一个请求在网络上传输时，事件循环可以切到另一个任务去执行。逻辑上看起来像是多个任务在同时推进，实际上是一个线程在快速切换，就像餐厅服务员同时给多桌客人点单，一桌在考虑时就去处理下一桌，而不是干等这一桌想完。

1.2 并发、并行、异步，是三个不同概念

这个区分很重要，因为面试里常被问，实际项目中理解偏了也会走错方向。

• 并行（Parallelism）：多核 CPU 同时执行多个计算任务，需要多个线程或进程，真正的同时进行。
• 并发（Concurrency）：多个任务在同一个时间段内交替执行，单个核心也可以实现，宏观上像是都在推进。
• 异步（Asynchrony）：一种并发的手段，不需要线程切换的开销，而是通过任务主动让出控制权来实现协作式调度。

Asyncio 属于第三种，是用户态的任务调度。它不解决计算密集问题的性能，也不是为了替代多线程。它的战场是 IO 密集，网络请求、文件读写、数据库查询这类场景。

这个表格不是绝对的，但在选型时它足够给你一个方向。CPU 密集型任务选多进程，琐碎 IO 任务优先考虑 Asyncio。

1.3 为什么是 Asyncio 而不是其他方案

Python 因为 GIL 的存在，多线程在处理 CPU 密集任务时并不能真正利用多核，所以很多人说“Python 多线程是假的”，这话不完全对——它只对 CPU 密集型任务是假的，对 IO 密集型任务，多线程一样有效。但线程切换有成本，而且线程一多，锁竞争、状态同步的复杂度会指数上升。

Asyncio 是在单线程里做任务切换，逻辑严谨、状态可控，而且语言层面的async/await语法已经非常成熟。它的缺点是写起来不如同步代码直白，但只要理解了事件循环，熟悉之后会发现它非常适合处理大量独立、轻量的 IO 任务。

我个人的体会是：如果你要并发发 1000 次 HTTP 请求，用线程池你会被线程管理和响应状态整得焦头烂额，用 Asyncio 加asyncio.gather()几行搞定。这也是 Asyncio 能成为 Python 标准库大杀器的原因。

2. 五个核心概念，一次理清

2.1 事件循环就是那个服务员

很多人第一次接触asyncio.run()到asyncio.get_event_loop()这些函数时会懵，觉得抽象。实际上，你只需要记住一个核心角色：事件循环（Event Loop）。

事件循环是一个不断运行的调度器。它维护一个待执行任务队列，按照规则依次取出一个任务开始执行。如果一个任务执行到await的位置，就把控制权交还给事件循环，事件循环继续去找下一个可以继续推进的任务。等到之前的 IO 操作完成，事件循环会把这个任务再次放回可执行列表。

整个程序就是在这样一个循环里运转。异步不是魔法，它只是把“等待”这个过程管理得井井有条。

2.2 协程是函数又不是函数

用async def定义的函数，调用它并不会立即执行函数体，而是返回一个协程对象。协程对象必须被事件循环驱动才能真正执行。

async def say_hello(): print("hello") await asyncio.sleep(1) print("world")

如果直接写say_hello()，你只会得到一个coroutine object，它不会执行。必须搭配await，或者交给asyncio.run()去执行，或者用asyncio.create_task()包装成任务。

这就像你写了一张任务卡，但没有服务员去翻牌子。事件循环就是那个翻牌子的服务员，await就是在告诉服务员“我允许你切走”。

2.3 awaitable 到底包括什么

能让await去等待的对象，我们称之为可等待对象。标准库里有三种：

• 协程对象（coroutine）
• 任务对象（Task）
• Future 对象（Future）

其中 Future 是最底层的概念，Task 是 Future 的子类，专门用来包装协程。日常开发里，大家接触最多的是协程对象和 Task，Future 通常在对库底层设计时才需要关心。

await一个协程，是等这个协程执行完；await一个 Task，是等这个任务执行完。它们的差别在于协程是“裸的”，Task 是有调度状态的。

2.4 Task：让多个协程并行推进

只写await coro()是串行执行，写成下面这样就变成了并发调度：

task1 = asyncio.create_task(fetch_data(1)) task2 = asyncio.create_task(fetch_data(2)) await task1 await task2

create_task()会立刻把协程包装成 Task 并注册进事件循环。注意，是“立刻注册”，不是“立刻执行”。它只是排队候场，事件循环会在合适的时机切换执行它。

如果不想自己管理 Task，还有两个常用工具：

• asyncio.gather()：并发执行多个可等待对象，返回所有结果，顺序与传入顺序一致
• asyncio.wait()：更灵活，可以控制返回条件和超时，返回 done 和 pending 两个集合

实际业务中gather用得多，因为它的 API 简单，结果好拿。wait适合处理需要分批获取结果的场景。

2.5 asyncio.run：程序的统一入口

为了让程序员只用关心业务而不去管理事件循环的创建和关闭，Python 3.7 引入了asyncio.run()。它会自动完成三件事：创建事件循环、运行传入的协程直到结束、关闭事件循环。

asyncio.run(main())

特别注意：asyncio.run()不能在一个正在运行的事件循环内部调用，也不能重复调用同一个事件循环。这在 Jupyter Notebook 里尤其容易踩坑，因为 Notebook 本身可能已经维护了一个事件循环。

如果你不是在写一个独立脚本，而是在已经运行的事件循环里继续添加任务，就不该用asyncio.run()，而是用loop.create_task()或asyncio.create_task()。

3. 从同步到异步：第一段可用代码

3.1 环境准备：Python 3.7 以上是基础

Asyncio 在 Python 3.4 就进入了标准库，但直到 3.7 的asyncio.run()和 3.8 的asyncio.create_task()才真正让异步代码变得顺手。所以，我的建议是：个人电脑上最低使用 Python 3.8，最好用 3.10 或更高版本。

为什么推荐高版本？因为 3.10 以后asyncio引入了一些性能优化和更清晰的错误提示，比如TaskGroup在 3.11 已经稳定。如果你在旧版本里学的写法，换到新版本大概率不会错，但新版本的确更好用。

3.2 先写一个最基础的协程

不啰嗦，直接上代码。

import asyncio import time async def task(name, seconds): print(f"开始任务：{name}") await asyncio.sleep(seconds) print(f"完成任务：{name}") return f"{name} 的结果" async def main(): start = time.perf_counter() result1 = await task("A", 2) result2 = await task("B", 2) print(f"串行耗时：{time.perf_counter() - start:.2f}s") print(result1, result2) asyncio.run(main())

这段代码用时大约 4 秒，因为它是串行地在等待。await task("A")发出后在等 2 秒，等完再执行task("B")。虽然两个任务本身没有依赖关系，写法却是按顺序处理的。

改成并发也很简单：

async def main(): start = time.perf_counter() results = await asyncio.gather(task("A", 2), task("B", 2)) print(f"并发耗时：{time.perf_counter() - start:.2f}s") print(results) asyncio.run(main())

同样是两个 2 秒任务，第二次耗时只有 2 秒左右。这就是异步的核心价值：把等待的时间重叠起来。

3.3 create_task 和 gather 怎么选

gather适合“一批任务一起等结果”的场景，它会挂起等待所有任务完成。但如果你希望先启动任务，然后继续做别的事，之后再统一收结果，create_task更合适。

async def main(): task_a = asyncio.create_task(task("A", 2)) task_b = asyncio.create_task(task("B", 3)) print("任务已启动，先做点别的事") await asyncio.sleep(0.5) print("别的事做完了，开始等结果") result_a = await task_a result_b = await task_b print(result_a, result_b)

这两种写法的区别：gather是创建后立刻进入等待状态，create_task是创建任务后代码可以继续往下走。需要把任务“发射出去”的时候就选create_task；只想一次性并行收集结果，就用gather。

3.4 asyncio.sleep 的真实身份

很多人看示例代码会下意识以为asyncio.sleep(2)就是time.sleep(2)的异步版。它俩的核心行为正好相反：time.sleep是阻塞整个线程，asyncio.sleep是假装睡 2 秒，但把控制权交还给事件循环。事件循环在这 2 秒内会去跑其他任务。

想验证这一点很简单：在asyncio.sleep(2)前后的print之间，再塞十几个小任务，那些小任务几乎全都能跑完。因为sleep在等的时间里，事件循环转向执行其他就绪的协程。

这让我每次写异步并发任务时都多一层意识：任务主动让出 CPU 的时机很关键，协程在await处让出，事件循环才能发挥作用。如果一个协程内部从头到尾没有await，它就是不可切出的，事件循环也没办法。

4. 真实场景：并发 HTTP 请求落地

4.1 用 aiohttp 而不是 requests

requests是同步库，如果在一个协程里直接调用requests.get()，这个调用会阻塞事件循环。就算外层用了 Asyncio，阻塞一次，全局任务全部卡住，这就完全失去了异步的意义。

异步 HTTP 客户端常用两个选择：aiohttp和httpx。httpx是近年越来越流行的库，它支持同步Client和异步AsyncClient，API 风格和requests比较接近，迁移成本低。这里以aiohttp为例，因为它稳定性好、踩过的坑少。

import aiohttp import asyncio async def fetch(session, url): async with session.get(url) as resp: return await resp.json() async def main(): urls = [ "https://api.example.com/data/1", "https://api.example.com/data/2", "https://api.example.com/data/3", ] async with aiohttp.ClientSession() as session: tasks = [asyncio.create_task(fetch(session, url)) for url in urls] results = await asyncio.gather(*tasks, return_exceptions=True) for url, result in zip(urls, results): print(url, result) asyncio.run(main())

注意ClientSession需要复用同一个实例，不要每发一次请求就创建一个ClientSession，那样会反复创建连接池，性能反而差。Session 内部的连接复用能大幅减少 TCP 握手开销。

4.2 控制并发度：信号量是靠谱的方式

一次性把几千个任务全部打进gather是不现实的，比如同时发 2000 个 HTTP 请求，可能直接把对端服务打挂，也可能让本地内存紧张。这时候要用信号量做并发限流。

async def fetch_with_limit(session, semaphore, url): async with semaphore: async with session.get(url) as resp: return await resp.text() async def main(): semaphore = asyncio.Semaphore(20) # 同时最多20个请求 async with aiohttp.ClientSession() as session: tasks = [fetch_with_limit(session, semaphore, url) for url in urls] results = await asyncio.gather(*tasks, return_exceptions=True)

信号量的原理简单，就是维护一个计数器：进入async with semaphore之后计数减 1；离开后计数加 1。20 是常用的经验值，具体数量取决于你的目标服务和机器资源，我一般从 20 起步，压测后上调或下调。

4.3 超时和重试，比想象中更重要

在异步高并发场景，一个一直没有响应的请求会占着 Task 不放。如果不设置超时，await response.text()可能挂到天荒地老。

推荐使用asyncio.wait_for：

async def fetch_with_timeout(session, url, timeout=10): async with session.get(url) as resp: return await asyncio.wait_for(resp.text(), timeout)

wait_for会在超时后立刻抛出asyncio.TimeoutError。重试逻辑可以配合asyncio.sleep实现：

async def fetch_with_retry(session, url, retries=3): for attempt in range(retries): try: async with session.get(url) as resp: return await resp.json() except (aiohttp.ClientError, asyncio.TimeoutError) as exc: if attempt == retries - 1: raise await asyncio.sleep(2 ** attempt) # 指数退避

指数退避在重试领域是常识级做法，第一次失败等 1 秒、第二次等 2 秒、第三次等 4 秒，避免重试风暴。

4.4 结果聚合与异常处理

gather有一个关键参数return_exceptions。默认情况下它是False，只要其中任意一个任务抛出异常，整个gather会立刻抛出异常，后面的结果也不给你。不设置return_exceptions=True时，还可能看到某个任务抛出的异常直接消亡，控制台只留下一句不痛不痒的警告。

设成True后，异常会作为返回结果中的一个值，塞在结果的相应位置。你挨个判断，是异常就记录日志，是结果就正常处理。

results = await asyncio.gather(*tasks, return_exceptions=True) for url, result in zip(urls, results): if isinstance(result, Exception): print(f"{url} 失败：{result!r}") else: print(f"{url} 成功：{result}")

这样做的好处是主流程不会被单个失败拖垮，坏处是你必须自己保证异常被消费，否则错误会静默丢失。实际项目里，我甚至会把这些失败请求的元数据写进一个回填队列，等主流程跑完再做一轮补偿式重试。

5. 混合负载：在线程池里跑同步代码

5.1 一个协程里绝对不能调同步阻塞函数

这是新手最容易犯的错误。一个增删改查的后端服务，其中某个函数依赖一个老旧的同步库，你为了性能把它包在协程里调，结果整个事件循环被这个同步调用堵住，其他所有协程全被拖慢。

比如：

async def bad_example(): requests.get("https://api.example.com/slow") # 阻塞！

这段代码看着在异步函数里，但requests.get()内部是同步 socket 调用，会直接阻塞当前线程。因为协程是在事件循环所在线程上运行的，阻塞这个线程等于阻塞一切。

正确的处理方式是，把这种同步阻塞调用丢给线程池去执行：

import asyncio import requests def sync_fetch(url): resp = requests.get(url) return resp.json() async def good_example(): loop = asyncio.get_running_loop() result = await loop.run_in_executor(None, sync_fetch, "https://api.example.com/data") return result

run_in_executor(None, func, *args)表示使用默认的线程池执行器，函数会被放到一个新线程里执行，协程则通过await等待结果。这样事件循环就不会被同步调用卡住。

5.2 进程池处理 CPU 密集的补充场景

如果某个任务既需要 IO，又包含大量 CPU 计算，把它全部放进事件循环会把同一线程的 CPU 占用打满，因为事件循环的调度并没有抢占机制。遇到这种情况，通常的做法是把重计算部分用run_in_executor配合ProcessPoolExecutor丢到子进程，进程间用队列传参，子进程算完再把结果投递回来。

from concurrent.futures import ProcessPoolExecutor async def compute_with_process(data): loop = asyncio.get_running_loop() with ProcessPoolExecutor() as pool: return await loop.run_in_executor(pool, compute, data)

不过进程开销大，适合计算量很大的场景。如果计算量本身就小，直接写同步代码就好，过度设计就是给自己挖坑。

5.3 线程安全的一次提醒

Asyncio 是单线程模型，这很好，但当你同时使用了线程池的run_in_executor，就要注意：不要在子线程里直接改全局字典或列表，否则同样会有竞态问题。子线程往主线程传数据时，最好只通过await拿返回值，不要在子线程里去操作主线程持有的对象。

还有一点比较隐蔽：asyncio的几乎所有公开 API 都不是线程安全的。如果你在子线程里调用了loop.create_task()或者asyncio.run_coroutine_threadsafe()混用方式不当，结果会非常难排查，报错也往往是突发性的。

6. 踩坑记录：那些年 Asyncio 教我的事

6.1 Task exception was never retrieved

这是 Asyncio 里最经典的一个问题。你在create_task里创建了一个任务，但后续没有任何地方await它，如果这个任务内部抛了异常，Python 会在垃圾回收时把这个异常再次抛出来，提示你“Task exception was never retrieved”。

我碰到过的典型场景是：在后台任务里查询数据库，查询失败抛了异常，但因为后台任务没人等，异常一直没人管，系统日志也不打印。直到程序退出或者任务被 GC，才突然冒出警告。

解决方案有两种。一种是所有create_task的返回值都纳入管理，要么await，要么用add_done_callback处理异常：

task = asyncio.create_task(background_job()) task.add_done_callback(handle_completion)

另一种是不把异常留给 Task 自己消化，而是在协程内部就try/except包住，保证异常不会逃逸。

6.2 RuntimeError: asyncio.run() cannot be called from a running event loop

在 Jupyter、某些测试框架里运行时，很容易触发这个问题。因为当前环境已经存在一个运行中的事件循环，你在代码里又调了asyncio.run()，这就相当于在一个正在招待客人的餐厅里，又开了一个新餐厅，显然不行。

解决方式是检测是否已有运行事件循环，有就复用：

try: loop = asyncio.get_running_loop() except RuntimeError: # 没有运行中的事件循环，可以asyncio.run asyncio.run(main())

但更好的方式是让代码结构适配环境。比如写一个入口函数，在独立脚本和交互环境里都能跑：

def entrypoint(coro): try: asyncio.get_running_loop() except RuntimeError: asyncio.run(coro) else: # 已经运行在事件循环里，直接创建任务 return asyncio.create_task(coro)

6.3 事件循环被同步库阻塞，任务集体延迟

我之前把一个基于pymysql的同步fetch_all调用直接放在了异步路由里，集成测试时发现接口响应从 200ms 暴涨到 3 秒。原因是并发量大的时候，同步调用把事件循环一堵，所有异步任务全都在排队等待。

排查过程：先打印每个协程开始和结束的时间戳，发现所有任务都卡在同一个函数附近；再用cProfile定位到这个同步调用占用了大量墙钟时间。最后改成run_in_executor后，接口恢复到了 200ms 量级。

这个坑的本质是：事件循环是协作式调度，它不会主动抢占一个正在运行的协程。协程自己不会让出 CPU，整个事件循环就跟阻塞了没两样。遇到“异步接口突然变慢，任务全部延迟”的问题，先别怀疑asyncio本身，先查有没有同步阻塞调用混在里面。

6.4 日志和调试开 DEBUG 模式

开发阶段建议开启 Asyncio 的 debug 模式，它能帮你发现很多潜在问题，比如事件循环被阻塞超过 100ms、回调执行时间过长等。

asyncio.run(main(), debug=True)

或者通过环境变量：

PYTHONASYNCIODEBUG=1 python app.py

开启后，如果某个协程执行时间过长，事件循环会打印警告，直接指出问题所在。我一般是在写并发逻辑第一天就开 debug，正式上线后再关掉，避免对性能产生额外影响。

6.5 任务取消与超时的正确处理

asyncio.wait_for或者cancel()都会给协程注入CancelledError。这个异常在except Exception里捕获不到，因为它继承自BaseException。如果你写了：

try: await do_something() except Exception: pass

你是接不住取消信号的，取消信号会被忽略，任务可能不会真正终止。

正确的清理方式是用except asyncio.CancelledError，做完必要清理后再重新抛出：

try: await long_running_task() except asyncio.CancelledError: # 做清理 session.close() raise

另外，TaskGroup（3.11+）对取消处理更友好，它会等待所有任务真正取消后才返回，而不是像某些旧写法那样取消后就没了下文。有条件的话建议优先用TaskGroup。

7. 从入门到顺手，还有几件事

如果你已经能写出并发 HTTP 抓取、后台定时任务、消息轮询这类代码，那 Asyncio 就已经不是入门阶段了。接下来我建议你关注几个再往上的点。

一是asyncio.Queue，它和queue.Queue的用法类似，但专门在协程里使用。生产者和消费者协程之间基于队列协作，特别适合做爬虫任务分发、消息批量消费。

q = asyncio.Queue(maxsize=20) async def producer(): for i in range(100): await q.put(i) async def consumer(): while True: item = await q.get() await process(item) q.task_done()

二是asyncio.StreamReader/StreamWriter，这是标准库提供的底层 TCP 接口，可以用来手写网络协议。想真正理解 Asyncio 的事件循环怎么驱动网络 IO，用这个手写一个简单客户端是很好的练手项目。

三是uvloop，它是一个把事件循环换成 Cython 实现的高性能替代库，API 和内置asyncio基本兼容。uvloop.install()一行接入，在高并发网络服务里能带来明显的性能提升。不过它依赖 libuv，装的时候留意一下平台兼容性。

最后唠叨一句环境层面的问题：如果你的业务并发量大，日志尽量异步写。用同步文件写入日志，在高并发事件循环里依然可能变成阻塞点，导致吞吐量上不去。常见的做法是用logging的标准库配一个独立线程的QueueHandler和QueueListener，日志写入放到后台线程，主流程完全不被阻塞。

我个人在实际项目里的习惯是：默认用asyncio处理所有 IO 密集型并发任务，但绝不排斥在局部细节上用同步库加线程池，一条路走到底并不是成熟的做法。关键是搞清楚“谁在等、谁在跑、谁的阻塞会影响谁”。

如果是刚接触异步，我建议你从今天这篇里的最小案例开始，先在你的机器上把gather、create_task、Semaphore这几个 API 各自跑一遍，再对照这些坑去逐个踩一踩、查一查。踩坑本身不可怕，怕的是不知道坑在哪里。把这几个点都过一遍，你对 Asyncio 的理解就已经超越了相当一部分网上所谓“手把手教程”的水平。