Python异步编程实战：掌握asyncio事件循环、并发控制与避坑指南

Python 异步编程，在实战里经常被两种极端态度对待：一种认为它是性能银弹，所有接口套上 async 就能吞吐翻倍；另一种觉得它太复杂、坑太多，不如老老实实写同步。我两边都待过，真实的结论是：异步是一个非常有用的工具，但前提是你得理解它到底在解决什么问题，以及它的代价在哪里。

这篇文章不打算重复官方文档，而是从实战视角串一遍。你会看到什么时候该用异步、事件循环和任务到底怎么协作、如何把同步代码改成异步、并发上限怎么控制、阻塞库怎么处理，以及我踩过的一些坑和排查思路。无论你是刚接触 asyncio，还是已经在项目里用了但总感觉哪里不对，都应该能找到一些有用的东西。

先说一个最常见的误区：很多人以为“用了异步代码就一定会更快”。这个认知在绝大多数场景下是错的。异步真正改变的不是单次操作的耗时，而是让大量等待时间重叠在一起。搞清楚这一点，后面的路就好走了。

1. 先别急着写代码：异步适合解决什么类型的问题

1.1 大部分性能问题不在“算”，而在“等”

如果你发现某个程序跑得很慢，第一步不是往上堆异步，而是先搞清楚时间到底花在哪。用 cProfile 或简单打点统计一下每个函数的耗时分布，你会发现大概率有超过一半的时间耗在网络请求、数据库查询、文件读写这类 IO 操作上。

拿一个很常见的批量查询场景来说。程序要循环调用某个内部接口，每次请求返回只要几十毫秒，但同步写法下，10 个请求就得等 10 个几十毫秒，串行叠加起来就变成了几秒。这时候 CPU 在干嘛？什么都没干，就在原地等内核把网络数据包送过来。这种“等待”就是异步编程要解决的核心问题——把不互相依赖的等待过程并发化。

我习惯用一个生活化类比来解释：烧水的时候你不会站在灶台边干等到水开，而是会去切菜、备料、煮饭。哪个环节需要等待，就先把它挂起来，转头去处理别的事情。asyncio 就是那个帮你调度“烧水、切菜、煮饭”的管理员。

1.2 为什么 CPU 密集任务别硬上 asyncio

和 IO 密集相对的是 CPU 密集任务，比如大量循环计算、字符串处理、编解码操作。这类任务的特点是没有等待，CPU 一直在干活。你把这种任务写成协程，切换过来换过去，不仅不能并行，反而要付出协程切换的开销，跑得比同步还慢。

更深一层的原因是 GIL。在 CPython 里，同一时刻只能有一个线程执行 Python 字节码，协程又是跑在线程上的，所以多个协程之间本质是轮流执行，而不是同时执行。想让 CPU 密集任务真正利用多核，正确做法是多进程，或者把计算核心交给 numpy、C 扩展这类能释放 GIL 的实现。

我在项目里一般直接用下面这个表做决策：

1.3 异步不是“加速魔法”：没有并发量就没有收益

还有一个反直觉的点：如果程序本身只有一个任务，比如一个请求一个响应，这时候用异步没有任何收益，甚至因为事件循环的调度开销会慢一点点。异步的优势体现在“单位时间内有很多个互不依赖的等待型任务”时。

所以动手改异步之前，先问自己三个问题：

1. 当前程序里是否存在大量同时发生的 IO 等待？
2. 这些等待之间是否彼此独立？
3. 如果用并发方式执行，会不会打爆下游系统或本机资源？

第三个问题很多人会忽略，后面讲到信号量的时候再展开。总之，判断该不该上异步，比怎么写异步更重要。

2. 理解 asyncio 的底层行为：事件循环、协程与任务

2.1 事件循环不是“循环”，而是一个调度中枢

asyncio 的核心是事件循环（event loop），你可以把它理解成一个协作式调度器。它会维护两个东西：准备执行的任务队列，和正在等待 IO 完成的任务列表。当某个协程执行到 await 时，它会把控制权交还给事件循环，事件循环接着去执行其他准备好的任务；等到那个 IO 事件真正完成了，事件循环再把对应的协程从暂停点恢复。

这个过程和操作系统的抢占式调度完全不同。线程是操作系统随时可以在任何指令位置打断你，协程则必须你自己主动让出控制权。所以异步代码里如果有一个地方不按规矩来，一直占着事件循环不放，那所有其他协程都会跟着遭殃。这就是为什么后面要专门讲阻塞库的处理。

2.2 调用 async def 函数并不会执行函数体

这个点几乎每个新手都错过。你看一个 async def 函数，以为调用一下就会跑，实际上它只是返回了一个协程对象，函数体一行都没执行。给你看个最直观的例子：

import asyncio async def do_work(): print("函数体执行了") await asyncio.sleep(1) async def main(): do_work() # 这里只是创建协程对象，函数体不会执行 await asyncio.sleep(1) asyncio.run(main())

运行之后你会发现函数体执行了根本没打印出来，反而可能收到一条coroutine 'do_work' was never awaited的警告。协程对象就像一份冷冻的菜谱，必须交给事件循环去“解冻”执行。要执行它，要么直接await，要么用asyncio.create_task包装成任务。

2.3 Task：让协程获得并发能力的载体

多个协程对象放在那里，它们自己是不会并发的。真正让协程进入事件循环调度队列的，是任务（Task）。asyncio.create_task的作用，就是把一个协程包装成 Task，注册进事件循环的调度列表，这样事件循环才有机会在某个协程 await 挂起时，切去跑另一个 Task。

可以用最早那个“烧水”类比：协程本身是做菜的步骤，Task 是你同时开了几个灶头，事件循环是那个来回看火、及时翻菜的大厨。只开一个灶头的话，一次只能做一道菜；开十个灶头，菜虽然还是那十个，但总时间能节省一大截。

Task 是 Future 的子类，Future 可以理解成一个“以后会放结果进去的盒子”。和 Future 打交道的情况并不多，日常开发中你主要会操作协程对象和 Task。需要注意的是，create_task必须在事件循环运行期间调用，所以它一般出现在 async 函数内部，而不是模块顶层。

2.4 await 本质：把控制权交出去，然后等结果

很多人把await理解成“等待结果返回”，这不算错，但它忽略了最重要的一层含义：await的同时，当前协程会挂起，控制权会交还给事件循环。事件循环看到别的任务准备好了，就会切过去执行，所以整个程序的并发才能运转起来。

await后面可以跟协程对象、Future 或者 Task。如果跟的是协程对象，事件循环会先把协程包成一个 Task 再调度；如果跟的是 Task，那就是等待这个具体任务完成。一次异步 IO 的完整链路大致是：协程 A await 一个网络请求，事件循环注册一个回调，然后调度协程 B；等网络请求的 socket 可读了，事件循环再从 A 的暂停点接着执行。

3. 把一段同步代码真正改成异步：完整改造过程

3.1 先看同步版代码长什么样

假设我们要批量请求一批接口地址，同步写法通常长这样：

import time import requests def fetch(url): resp = requests.get(url, timeout=5) return resp.status_code def main(): urls = [...] # 假设有 50 个接口地址 start = time.perf_counter() results = [] for url in urls: results.append(fetch(url)) elapsed = time.perf_counter() - start print(f"耗时: {elapsed:.2f}s, 结果数: {len(results)}") if __name__ == "__main__": main()

这段代码的问题就在for循环里。每次调用fetch(url)，程序都会发起一次网络请求，然后阻塞在那里等服务器响应，等完了才去请求下一个。请求之间明明没有任何依赖关系，完全应该同时发出去。

3.2 换成异步客户端，并解释为什么快

改造成异步，需要一个支持异步的 HTTP 客户端。我个人常用httpx.AsyncClient，如果你项目里用的是aiohttp也一样，核心思路不变：

import asyncio import httpx async def fetch(client, url): resp = await client.get(url, timeout=5) return resp.status_code async def main(): urls = [...] # 同样的 50 个接口地址 async with httpx.AsyncClient() as client: tasks = [asyncio.create_task(fetch(client, url)) for url in urls] results = await asyncio.gather(*tasks) print(f"结果数: {len(results)}") if __name__ == "__main__": asyncio.run(main())

这段代码里有两个关键变化。第一，await client.get(...)会真正发起异步请求，在等待响应的时候把控制权交回事件循环；第二，asyncio.create_task把 50 个fetch协程全部丢进调度队列，事件循环会不断切换执行那些已经收到响应的任务，而不是呆等某一个。

为什么AsyncClient比短连接版快得多，还有一个容易被忽略的点：连接复用。async with httpx.AsyncClient()创建的连接池会复用底层的 TCP 连接，大量请求可以共享同一批连接，省掉了反复握手的时间。这一点在改造成异步时往往是隐性收益，别只盯着并发模型看。

3.3 数据库和文件 IO 也要跟着异步化

HTTP 请求只是异步化的一部分。项目里如果有数据库操作，用requests的同步习惯同样会拖后腿。SQLite 场景推荐aiosqlite，PostgreSQL 场景推荐asyncpg或 SQLAlchemy 的 async 版本。文件读写也有aiofiles这样的库。

不过我不建议一股脑把整个项目全部异步化。如果某块功能只是偶尔读一个小文件、写一条日志，同步执行也就几百微秒，硬改成异步反而增加维护成本。异步的收益是放大“等待时间占比高”的场景，低频小 IO 不值得折腾。判断标准还是回到第一节那个问题：这个操作是否在大量等待？

还有一点必须提醒：异步是有“传染性”的。一旦你把某个底层 IO 函数改成了async def，所有直接或间接调用它的函数，要么也变成async并await它，要么就得用别的办法去调它。改之前要对自己的调用链有个全貌，否则会陷入改一层崩一层的窘境。

4. 并发不是越大越好：信号量、批量任务与超时

4.1 不加限制的并发会搞崩系统和被对方限流

很多人看到异步能并发，第一反应是把任务全部create_task，用gather一捞就完事。你试试一次发 500 个请求，结果大概率是一部分请求超时，一部分连接被拒，甚至本机文件描述符被打满。上游服务如果有连接数限制或者限流策略，下一秒你的 IP 就可能被拉黑。

并发不是免费的。每个并发任务都占用内存、文件描述符、连接池里的连接，还会给下游系统带去压力。所以真实项目里几乎一定会做的一件事，就是控制并发上限。

4.2 Semaphore 是控制并发最简单的手段

asyncio.Semaphore就是一个并发闸门，初始化时设置允许的最大并发数，每次任务执行前获取一个信号量，执行完释放。超过上限的协程会排队等待。示例：

import asyncio import httpx async def fetch_with_limit(client, semaphore, url): async with semaphore: resp = await client.get(url, timeout=5) return resp.status_code async def main(): urls = [...] # 100 个地址 semaphore = asyncio.Semaphore(20) # 最多 20 个并发 async with httpx.AsyncClient() as client: tasks = [ asyncio.create_task(fetch_with_limit(client, semaphore, url)) for url in urls ] results = await asyncio.gather(*tasks) print(f"结果数: {len(results)}") asyncio.run(main())

注意信号量要在事件循环内部创建，不要在模块顶层创建后再拿进不同的事件循环用。不同事件循环之间共享同一个信号量，在 Python 3.10 之后可能不会立刻报错，但行为是不可预期的，后面踩坑部分会细说。

4.3 并发上限不是拍脑袋定的，要压一压

我见过的团队，并发数要么写成 10，要么写成 1000，但很少有人真去压测验证。比较靠谱的做法是：在测试环境模拟目标接口的特征，分别用不同并发度跑一轮，把耗时、失败率、内存占用记录下来。

我在某个模拟接口上跑过一轮趋势数据，大概是这样的（具体数值和机器配置强相关，看趋势就行）：

这张表很有代表性：并发从 10 提到 50，收益明显；再往上翻，失败率和内存开始恶化，总耗时甚至不降反升。所以并发上限不要拍脑袋定 1000，先压一轮再说。

4.4 gather、wait、TaskGroup 怎么选

批量任务编排时，asyncio.gather是最常用的，但并不是唯一选择。我整理一下区别，方便你按场景选：

我自己现在新代码优先用TaskGroup，因为它更安全。看这个示例：

async def main(): async with asyncio.TaskGroup() as tg: task1 = tg.create_task(fetch(client, url1)) task2 = tg.create_task(fetch(client, url2)) # 到这一行，里面的任务必然全部执行完了

TaskGroup进入async with时会等待所有子任务结束，中途如果某个任务抛异常，它会自动取消剩下的兄弟任务，然后统一把异常抛出来。这种结构化语义能避免一个任务挂了、其他任务还在后台乱跑的问题。

4.5 超时：没有超时的并发就是定时炸弹

并发任务一旦没有超时限制，整个程序可能被某个“卡死”的请求拖住。最简单的做法是用asyncio.wait_for：

import asyncio async def fetch_with_timeout(client, url): try: result = await asyncio.wait_for(client.get(url, timeout=5), timeout=10) return result.status_code except asyncio.TimeoutError: return None

Python 3.11 之后还引入了asyncio.timeout上下文管理器，配合async with更直观。注意：任务超时触发后，wait_for会尝试取消内部协程，如果协程里开着连接之类的资源，记得处理asyncio.CancelledError释放资源。

5. 事件循环里最怕的事：遇到同步阻塞库怎么办

5.1 一个 time.sleep 如何拖垮整个世界

很多人写了 async 函数，但里面调用的却是同步库，比如requests.get、time.sleep、旧版文件读写库。这会导致事件循环被整个阻塞，别指望其他协程还能继续跑。给我一个很小的演示：

import asyncio import time async def worker(name): while True: print(f"{name} 工作了一轮") await asyncio.sleep(1) async def main(): t = asyncio.create_task(worker("A")) time.sleep(3) # 同步阻塞 3 秒 await asyncio.sleep(3) asyncio.run(main())

运行后你会看到worker在time.sleep(3)期间完全没有执行机会，因为事件循环线程已经被time.sleep霸占了。哪怕只是 100 毫秒的同步阻塞，在高并发下也会造成连锁延迟。

5.2 用 asyncio.to_thread 把阻塞调用挪出去

如果某个同步函数实在没法改成异步，比如市面上的老三方库只有同步版本，可以用asyncio.to_thread把它丢到线程池里跑：

import asyncio import time def old_sync_request(url): time.sleep(2) return f"结果: {url}" async def main(): result = await asyncio.to_thread(old_sync_request, "https://example.com") print(result) asyncio.run(main())

asyncio.to_thread本质上会在线程池中执行同步函数，再把结果通过一个 Future 桥接回来。对于文件操作、DNS 查询、老旧的同步网络库这类阻塞型 IO，效果立竿见影。但要泼一盆冷水：如果这个“阻塞函数”是高强度 CPU 计算，to_thread并不会有太大帮助，因为 GIL 的存在，多个 Python 线程依然不能同时执行 Python 字节码。

5.3 CPU 密集部分交给多进程

处理 CPU 密集任务的正确姿势，是用ProcessPoolExecutor配合loop.run_in_executor：

import asyncio from concurrent.futures import ProcessPoolExecutor def heavy_calc(n): return sum(i * i for i in range(n)) async def main(): loop = asyncio.get_running_loop() with ProcessPoolExecutor(max_workers=4) as pool: result = await loop.run_in_executor(pool, heavy_calc, 10_000_000) print(result) asyncio.run(main())

这里要注意，传给子进程的函数和参数必须能被 pickle 序列化，返回值也一样。子进程开销比线程大不少，所以任务量太小的话，反而会被进程创建成本吃掉收益。实际工程里，我一般会把一个比较大的计算批次拆成多个中等大小的任务丢给进程池，而不是一个一个零碎地派发。

6. 生产环境中踩过的坑：四条排查链路复盘

6.1 协程从未被 await：你以为并发，其实没跑

这个现象太常见了。代码看起来没问题，运行起来目标任务一个都没执行，只留下一条coroutine ... was never awaited的警告。问题通常出在忘写await，或者把一个协程函数当普通函数调用。

排查思路很简单：看到警告先定位协程名字，顺着调用链找到那一行，补上await或改成create_task。更省事的方法是在起项目时直接开 asyncio 调试模式：

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

调试模式下，asyncio 会输出更多未完成任务和慢回调的日志，很多“幽灵任务”都能现形。

6.2 Task 被垃圾回收：任务悄然消失

另一个容易忽略的坑是create_task的返回值没有保存引用。官方文档明确建议保留任务的引用，否则任务可能在执行过程中被垃圾回收掉，表现就是任务只跑了一半甚至从未开始。事件循环内部持有的引用不足以保证任务不被回收。

我以前在一个定时任务改造里踩过这个坑：某批任务在低峰期正常，高峰期总是莫名少几个，后来发现create_task返回的任务对象没有被任何变量引用。修复方式很简单，用列表或集合把任务存起来：

tasks = set() for url in urls: task = asyncio.create_task(fetch(client, url)) tasks.add(task) task.add_done_callback(tasks.discard)

用set保存并在完成回调里把它移除，既保持引用，又不会造成任务集合无限膨胀。

6.3 定时任务突然不准时：用 py-spy 看线程堆栈

有一次我在维护某个定时同步服务，发现定时任务应该每 5 秒跑一次，但日志显示有时候 1 分钟才跑一次。第一反应是调度器出问题了，查了半天发现调度器正常，罪魁祸首是事件循环线程被某个同步库调用阻塞住了。

当时用的排查工具是py-spy，不需要重启进程，直接导出线程堆栈：

py-spy dump --pid <进程号>

堆栈里清清楚楚地看到事件循环线程卡在那个老同步网络库的connect等待上。替代方案是把日志放大，输出每次事件循环恢复的时间点，对比任务实际执行间隔；但py-spy能看到完整调用链，效率高得多。定位到根因后，把那个同步调用替换成异步版本，定时任务立刻恢复正常。

6.4 事件循环混用：asyncio.run 嵌套和外层循环冲突

还有一类坑，是在已经运行的事件循环里，又调用了asyncio.run()。比如这样：

import asyncio async def inner(): await asyncio.sleep(1) async def outer(): asyncio.run(inner()) # 报错 asyncio.run(outer())

运行会直接抛RuntimeError: asyncio.run() cannot be called from a running event loop。这是因为同一线程同一时刻只允许一个运行中的事件循环。类似问题还出现在跨循环复用资源上：在一个asyncio.run里创建的Semaphore、AsyncClient，拿到另一个事件循环的协程里用，轻则报attached to a different loop，重则数据错乱。所以资源对象要在同一个异步上下文里创建和销毁，别跨函数乱传。

6.5 开启调试模式，别等问题上线了再查

生产环境没法随时开 debug，所以在预发和本地测试时，建议环境变量PYTHONASYNCIODEBUG=1跑一遍。开启后 asyncio 会检查未等待的协程、未关闭的资源、执行时间过长的回调，这些信息在排查问题的时候特别好用。

还有一个习惯很重要：给任务命名。create_task(..., name="fetch-xxx")之后，日志里能直接看到是哪个任务出了问题，而不是看到一串匿名任务傻眼。尤其任务多的时候，这条经验能帮你省下大量排查时间。

7. 把异步写进工程：架构层面的几个习惯

7.1 异步要从底层开始，而不是外层包一层

异步化最怕的是“外层统一改 async，底层还是同步库”。这样的代码表面上是异步，实际请求还是阻塞的，事件循环该卡照样卡。正确思路是从 IO 边界开始改造：先把最底层的网络、数据库、文件操作换成异步库，然后调用链一路向上 await，最后在入口处统一聚合。

如果老项目很大，不要指望一天改完。我一般会从流量最集中、等待占比最高的模块下手，比如批量查询、消息消费这类场景，先把一小块链路异步化跑通，再逐步扩大边界。每次改造后对比耗时和失败率，不达标就回滚，用数据说话。

7.2 生命周期管理：连接池、优雅关闭与任务取消

真实服务里，异步资源要跟着进程生命周期走。服务启动时创建好AsyncClient、连接池、信号量，而不是每个请求临时创建；服务关闭时先停止接收新任务，再统一取消仍在运行的任务，给它们一个宽限期处理收尾。

任务取消处理也容易写错。task.cancel()会在任务内部抛出CancelledError，如果你在代码里盲目捕获所有异常，可能把取消信号给吞掉，导致任务退出不了。正确做法是捕获CancelledError后释放资源，然后继续向上抛出，让它结束协程。这个细节在服务的优雅下线里特别重要。

7.3 监控三个关键指标，比盯着业务日志更可靠

异步系统一旦出问题，业务日志往往是“看起来一切正常”。我一般在监控里盯三个指标：当前 pending 任务数量、事件循环调度延迟、连接池等待时间。pending 任务持续上涨，说明任务产出速度跟不上消费速度；事件循环调度延迟变大，说明有阻塞操作在抢占事件循环；连接池等待时间变长，则可能要对并发上限做调整。

这些指标并不难采集，核心思路是在事件循环里定期记录时间戳，或者在每个任务的前后打点，把数据送到监控系统里做趋势图。指标比经验靠谱，因为它们不会说谎。

最后分享一点个人体会。我真正把异步用明白，不是在学会async/await语法的那天，而是在某次线上事故里，用堆栈工具看到事件循环被同步调用卡住的那一刻。从那之后我每次写完异步代码，都会自查几个问题：有没有真正 await？任务有没有保留引用？并发有没有上限？动作有没有超时？日志能不能追溯每个任务？把这几个问题答清楚，一套异步代码基本就能稳定落地了。