一直用同步代码写业务总觉得瓶颈在网络请求上。后来把一批接口调用改成异步同样一批任务耗时从 6 秒压到 1.2 秒左右代码量没增加多少。当时就一个感觉异步编程这个东西真不是面试题里考考就算了的它是真能在生产环境里换回实打实性能的。这篇就围绕 async、await 和 asyncio 这三个核心聊聊我在实际操作中怎么用、怎么排坑、怎么设计任务调度希望能给正打算入坑或已经写异步但经常被坑到的朋友一些参考。先说清楚这篇适合谁你已经会 Python 基础语法知道函数、类、装饰器是什么但还没系统写过异步代码或者你写过一点 async/await但遇到“协程不执行”、“任务取消不掉”、“阻塞把事件循环卡死”这类问题想找到原理层面的解释。整个内容从概念到实践再到排查手段都会覆盖。1. 先把三件事分开async、await、asyncio 到底各管什么很多初学者把这三个词混在一起以为它们是同一个东西的三种写法这是最大的误解。实际它们各干各的配合起来才构成完整的异步编程体系。1.1 async把普通函数变成可暂停的函数在函数定义前加上asyncPython 就会把这个函数变成一个协程函数。调用它不会立刻执行函数体而是返回一个协程对象。这个协程对象只有被事件循环调度时才会真正运行函数体里的代码。async def fetch_data(): return {code: 200, data: ok} coro fetch_data() print(type(coro)) # class coroutine print(coro) # coroutine object fetch_data at 0x...你看到这里会疑惑为什么没有直接返回字典因为协程函数的设计目标就是“可挂起、可恢复”它不能在调用瞬间把整个函数体跑完否则和普通函数没有区别。async的本质是告诉 Python这个函数内部可能有需要等待的操作请允许它在等待时让出控制权。我遇到不少初学者直接写result fetch_data()然后拿到一个 coroutine 对象再去打印就报错。这是最典型的第一个坑。牢记协程函数必须被事件循环调度或者被另一个协程用await等待代码才会真正执行。1.2 await把控制权交出去的“暂停按钮”await只能用在协程函数内部它的作用是等待一个可等待对象awaitable。可等待对象包括协程对象、Future 对象、Task 对象以及实现了__await__方法的对象。import asyncio async def main(): print(开始) await asyncio.sleep(1) print(结束) asyncio.run(main())这里await asyncio.sleep(1)的意思是告诉事件循环“我要等 1 秒这 1 秒里你去做别的事吧”。事件循环收到这个消息后会把控制权交给其他准备就绪的协程。1 秒后事件循环回来继续执行这一行之后的代码。用一个生活化的类比你在餐厅点餐await等于告诉服务员“菜好了叫我我先刷手机”服务员接着去服务其他桌。如果不用await整桌菜必须一道一道等着上后面的桌全被堵死。await只能等待可等待对象不能等待普通的耗时的同步函数。比如await time.sleep(1)会直接报TypeError。必须用await asyncio.sleep(1)代替。这个“只等待可等待对象”的规则是语法强制性的也是很多新手写异步代码时容易卡住的地方。1.3 asyncio整体调度的大脑asyncio是 Python 标准库提供的事件循环实现。事件循环event loop是异步编程的核心引擎它维护一个任务队列不断循环检查哪些任务可以继续执行然后逐个推进它们。import asyncio async def task(name, delay): print(f{name} 开始) await asyncio.sleep(delay) print(f{name} 结束) async def main(): # 创建三个任务并发调度 t1 asyncio.create_task(task(A, 2)) t2 asyncio.create_task(task(B, 1)) t3 asyncio.create_task(task(C, 3)) # 等待所有任务完成 await asyncio.gather(t1, t2, t3) asyncio.run(main())事件循环的工作方式用一个简化流程来描述维护一个就绪队列存放可以继续执行的协程。取出一个协程执行到遇到await或协程结束。如果遇到await就把协程挂起注册一个回调等待事件发生后恢复。继续处理队列中的下一个协程。所有协程都结束后事件循环退出。有了事件循环asyncio才能做到“单线程内并发”。它不是多线程不涉及 GIL 竞争没有线程切换的开销也正因如此它能轻松创建成千上万个并发任务而不会把内存耗尽。2. 三种并发手段gather、wait 和 Task实际项目中怎么选真正写业务的时候你不会只 await 一个协程更多时候是要并发跑一批任务。asyncio 提供了好几个并发工具用不对会踩不少坑。2.1 asyncio.gather最省心的批量并发gather是最常用的批量并发函数传入多个可等待对象并发执行等全部完成后统一返回结果列表。它有一个我一直很依赖的特性返回顺序和传入顺序一致不管内部谁先完成。import asyncio async def fetch(url, delay): await asyncio.sleep(delay) return f结果来自 {url} async def main(): urls [a.com, b.com, c.com] result await asyncio.gather( fetch(urls[0], 3), fetch(urls[1], 1), fetch(urls[2], 2), ) print(result) # 顺序永远是 [结果来自 a.com, 结果来自 b.com, 结果来自 c.com] asyncio.run(main())这个特性在做“并发请求接口然后按固定顺序处理结果”时特别有用——你不需要自己维护一个 task 到结果的映射gather 返回的列表天然对齐输入顺序。gather的另一个重要特性是“默认一票否决”如果其中一个任务抛异常gather会立即抛出异常同时会取消其他还没完成的任务。这在某些场景下是你想要的一个失败了全部取消但有些场景你想让其他任务继续跑完那就需要配置return_exceptionsTrueasync def safe_run(): result await asyncio.gather( may_fail_task(), normal_task(), return_exceptionsTrue, # 异常不会抛出来而是作为返回值包含在结果里 )开启后失败任务的异常会作为返回值出现在结果列表中不会打断其他任务。读取结果时需要对每个返回值做类型判断如果它是Exception的子类实例说明这个任务失败了。2.2 asyncio.wait更贴近底层的控制wait与gather的区别在于它接受 Task 对象集合返回(done, pending)两个集合。你能精确知道哪些任务完成了哪些还没有配合timeout参数实现“限时等待”。import asyncio async def worker(name, delay): await asyncio.sleep(delay) return name async def main(): tasks [ asyncio.create_task(worker(A, 2)), asyncio.create_task(worker(B, 5)), asyncio.create_task(worker(C, 1)), ] done, pending await asyncio.wait(tasks, timeout3) for task in done: print(f已完成: {task.result()}) for task in pending: print(f未完成: {task}) asyncio.run(main())这里 3 秒超时后只有 A 和 C 在 done 集合里B 还在 pending 中。对这种场景gather就做不到这么精细。但用wait要格外注意timeout到了之后未完成的任务不会自动被取消它们仍然在事件循环里运行。如果你不再需要它们要手动调用task.cancel()否则这些任务会继续消耗资源甚至在你asyncio.run退出时收到警告信息。# 超时后主动取消未完成任务 for task in pending: task.cancel() await asyncio.gather(*pending, return_exceptionsTrue)我一开始写这行代码时漏了后面的gather结果CancelledError没有被处理程序直接打了一个 Python 原生 traceback。取消一个任务后必须等它真正结束否则它抛出的CancelledError就悬在那里。2.3 asyncio.create_task手动管理每个任务create_task是把一个协程包装成 Task然后丢进事件循环的调度队列里。用gather和wait的时候内部其实也是用create_task来包装协程的。import asyncio async def heart_beat(): for i in range(10): print(f心跳 {i}) await asyncio.sleep(1) async def main(): task asyncio.create_task(heart_beat()) # do something else await task # 确保任务执行完 asyncio.run(main())创建任务后如果不await它有两种结果如果事件循环在任务完成前就退出了任务会被直接丢弃并给出“Task was destroyed but it is pending”的警告如果主协程通过别的逻辑让事件循环保持了足够长的时间任务可能会在后台安静地运行但没有地方接收它的结果和异常。这个行为容易埋雷——异常在事件循环里裸奔。所以在使用create_task时我的习惯是把创建出来的 Task 对象统一保存到一个列表或字典里后续用gather统一收集结果或异常。2.4 工具选择的决策表我把三个工具的适用场景整理成一张表方便实际写代码时快速决策工具核心特性推荐场景注意事项gather返回结果与输入顺序一致并发接口请求、批量任务收集结果默认一个失败全取消需用return_exceptionsTrue控制wait返回 done/pending 集合支持 timeout限时任务淘汰、分布式协调类需求超时未完成任务需手动取消create_task最底层的任务创建方式需要单独控制任务生命周期、后台任务必须保存引用等待任务结束或捕获异常3. 从实际业务场景出发协程、线程和进程怎么搭配异步不是万能的有一种情况它完全帮不上忙——CPU 密集型任务。如果你在协程里写了大量循环计算事件循环会被卡死其他所有协程都得排队等待。这里得说清楚什么场景能用协程什么场景必须用线程或进程。3.1 I/O 密集与 CPU 密集的边界协程的核心优势是充分利用 I/O 等待时间。网络请求、文件读写、数据库查询、消息队列消费这些操作的共同特点是CPU 实际计算量很小大部分时间都在等外部系统返回。等待是协程的强项——等待时让出 CPU去做别的任务。但如果你的任务是纯计算比如解析超大 JSON、做图像处理、跑机器学习模型推理CPU 一直是满的。这时候用协程不仅没有提升效果反而会因为单线程只能用一个核而比多进程方案更慢。判断标准很简单任务花在等的时间占比高 - 协程任务占满 CPU - 多进程有阻塞调用还不想重写 - 多线程。3.2 在线程池中跑阻塞任务现实里你经常会遇到不得不调用旧的同步库的情况比如requests、time.sleep、数据库驱动。直接把同步函数丢进协程里执行会阻塞事件循环一个time.sleep(5)就会让所有并发任务全部卡住 5 秒。解决方案是为阻塞任务提供一个线程池或进程池的桥梁。asyncio 提供了三个现成的方法import asyncio import time async def main(): # 1. asyncio.to_thread: 把同步函数丢到默认线程池中执行 result await asyncio.to_thread(time.sleep, 2) print(完成) # 2. loop.run_in_executor: 更灵活可指定线程池或进程池 import concurrent.futures with concurrent.futures.ThreadPoolExecutor(max_workers4) as pool: result await asyncio.get_running_loop().run_in_executor(pool, time.sleep, 2) print(executor 完成) # 3. 新版推荐的 loop.to_thread 方法 loop asyncio.get_running_loop() result await loop.run_in_executor(None, time.sleep, 2) asyncio.run(main())asyncio.to_thread是 Python 3.9 以后新增的代码简洁、好理解底层就是run_in_executor(None, ...)的封装。我把requests阻塞调用改写成await asyncio.to_thread(requests.get, url)之后并发能力提升了一个量级而且不用引入第三方库。同时一定要记住写异步时尽量减少“异步转同步”的反模式。很多人会在主线程用asyncio.run()启动事件循环后又想拿task.result()于是加一遍loop.run_until_complete()结果在已有事件循环时调用它会报错。这种“协程外面包一层同步调用”的别扭写法应该被消灭。3.3 线程安全与队列同一资源的竞争问题协程是单线程的理论上不需要加锁。但如果你在协程里通过to_thread跑了多线程代码或者协程与多线程混用就要注意线程安全问题。asyncio 提供的Queue是线程安全的这一点我反复确认过。你可以放心地把生产者协程放入队列消费者协程从队列取数据。配合put_nowait、get等方法可以构建异步生产者-消费者模型import asyncio async def producer(queue, n): for i in range(n): await queue.put(fitem-{i}) await asyncio.sleep(0.1) await queue.put(None) # 发送结束信号 async def consumer(queue): while True: item await queue.get() if item is None: break print(f消费: {item}) queue.task_done() async def main(): queue asyncio.Queue(maxsize10) prod asyncio.create_task(producer(queue, 20)) cons asyncio.create_task(consumer(queue)) await prod await cons asyncio.run(main())生产-消费模式在爬虫、消息处理、日志收集等场景下非常常见。队列的maxsize参数可以用来限制内存占用防止生产者一下子塞太多数据。如果多个协程共享一个普通的非线程安全的对象比如一个普通的list在单线程事件循环内其实也不太安全——因为协程任务在await时可能会被打断逻辑上形成一个“时序差”。在这种情况下就需要用asyncio.Lock来保护临界区lock asyncio.Lock() async def safe_increment(counter): async with lock: temp counter[value] await asyncio.sleep(0) # 模拟被打断 counter[value] temp 1必须强调的是asyncio.Lock不是线程锁它只保护协程之间的互斥不能保护线程。如果多线程也访问同一个对象你需要搭配标准库threading.Lock或者直接规避共享可变状态。4. 核心细节Task 生命周期、取消机制和事件循环想真正得心应手需要理解 Task 的生命周期。任务不是天生就能跑完的它和外部事件、取消信号、异常之间都有复杂的交互。4.1 任务的三种结束方式正常返回、抛出异常、被取消一个 Task 最终只有三种结局协程执行到return任务正常完成task.result()返回结果。协程里抛出了异常任务状态是失败task.result()重新抛出这个异常。协程收到CancelledError任务被取消task.cancelled()返回True。被取消与抛异常是两种完全不同的机制。CancelledError继承自BaseException而不是Exception所以在协程里用except Exception是捕获不到取消信号的。要正确捕获async def graceful_task(): try: while True: await asyncio.sleep(1) print(工作中) except asyncio.CancelledError: print(收到取消做清理) raise # 重新抛出保持取消语义注意捕获到CancelledError之后你必须在清理结束后重新抛出它raise否则系统会认为这个任务已经成功完成了取消状态就被破坏了。这在很多官方文档和源码里都有体现是任务优雅关闭的关键。4.2 超时控制wait_for 与 async with timeout对于外部请求不能无休止等待。asyncio.wait_for提供了超时中断能力import asyncio async def slow_request(): await asyncio.sleep(10) return 太慢了 async def main(): try: result await asyncio.wait_for(slow_request(), timeout3) print(result) except asyncio.TimeoutError: print(超时了)超时后会取消被等待的协程。如果想在超时后仍然保留协程运行比如发个取消请求但可以继续记录日志那就需要更灵活的超时模式。Python 3.11 引入了asyncio.timeout()上下文管理器让超时控制的写法更优雅async def main(): try: async with asyncio.timeout(3): result await slow_request() except TimeoutError: print(超时了)asyncio.timeout(3)内部的实现更智能在asyncio.CancelledError和超时之间做了很好的隔离能够避免被外层except asyncio.CancelledError误捕获。如果项目还在用老版本我会推荐用wait_for它的行为在多年迭代后就相对稳定。4.3 事件循环的关闭asyncio.run 与手动管理asyncio.run()从 Python 3.7 开始成为首选入口。它做的三件事值得知道创建新的事件循环。运行传入的协程直到完成。关闭事件循环取消剩余任务清理资源。这三点保证了我们不需要自己关心循环生命周期。但是如果在asyncio.run内创建的 Task 没等它完成就想要从外部访问最常见的错误就是在asyncio.run之后去拿task.result()任务在事件循环关闭时已经被清掉了。如果你真的要长时间运行一个异步应用这里的建议是维护一个全局事件循环。但为了安全我倾向于用asyncio.run保持作用域隔离。业务足够简单不用担心。如果应用确实很复杂再去动loop asyncio.new_event_loop()也不迟但此刻必须自己处理 “关闭时取消所有任务” 的逻辑——这是一个很容易留下资源泄漏的点。4.4 三个容易误用的 asyncio API有几个方法是大家经常从直觉上拿来用、但实际不对的API直觉用法正确用法asyncio.sleep(0)无所谓常用于让出控制权让任务切换不阻塞事件循环loop.run_until_complete(coro)在已有事件循环时用会创建/获取事件循环并卡住它不能嵌套调用asyncio.get_event_loop()在任意线程直接用在协程内部必须用get_running_loop()否则在子线程中可能没有默认循环asyncio.sleep(0)我强烈建议学会使用。它本质上是“让出一次执行机会”可以解决很多长循环卡死事件循环的问题。比如async def heavy_loop(): for i in range(100000): if i % 1000 0: await asyncio.sleep(0) # 让事件循环处理其他任务 # do something没有这个sleep(0)每次await之外长时间占用 CPU整个异步体系就崩了。5. 实战一个完整的异步爬虫案例纸上谈兵不如完整跑一个例子。下面是一个小爬虫 demo目标是从多个 URL 抓取数据解析 JSON再写进本地文件中间加入限速和重试逻辑。这个例子把前面讲的 gather、wait_for、超时、任务取消都综合起来了。5.1 需求与整体设计假设要抓取 20 个接口数据每个接口响应时间在 1~3 秒之间波动有些接口可能偶尔超时。需要做以下几点并发发起 20 个 HTTP 请求不一个一个地等。每个请求设置 5 秒超时。请求失败或超时最多重试 2 次。全部完成后将结果统一写入本地 JSON 文件。这个场景是典型的协程并发业务。网络 I/O 等待时间占大头CPU 计算极少。如果用同步requests逐个请求总耗时接近所有请求耗时之和用协程并发总耗时约等于最长那个请求的时间。5.2 代码实现先用一个异步 HTTP 客户端。标准库urllib是同步的这里用aiohttp这是目前 Python 生态里最常用的异步 HTTP 库。import asyncio import aiohttp import json MAX_RETRIES 2 SIZE 20 async def fetch_with_retry(session, url, timeout5): 带重试和超时的请求函数 for attempt in range(MAX_RETRIES 1): try: async with session.get(url, timeouttimeout) as resp: if resp.status 200: return await resp.json() else: print(f请求 {url} 状态码 {resp.status}) except asyncio.TimeoutError: print(f请求 {url} 超时第 {attempt 1} 次重试) except aiohttp.ClientError as e: print(f请求 {url} 出错: {e}) await asyncio.sleep(0.5) # 重试前稍等 return None async def main(): urls [fhttps://api.example.com/data/{i} for i in range(SIZE)] results {} async with aiohttp.ClientSession() as session: tasks [asyncio.create_task(fetch_with_retry(session, url)) for url in urls] # 等待所有任务完成但不想因为一个失败取消所有任务 completed, pending await asyncio.wait(tasks, timeout30) # 超时未完成的任务取消掉 for task in pending: task.cancel() if pending: await asyncio.gather(*pending, return_exceptionsTrue) for task in completed: try: result task.result() if result is not None: results[task.get_name()] result except Exception as e: print(f任务异常: {e}) with open(data.json, w, encodingutf-8) as f: json.dump(results, f, ensure_asciiFalse, indent2) print(f完成保存 {len(results)} 条记录) asyncio.run(main())这段代码的结构很适合理解任务集合管理。asyncio.wait(tasks, timeout30)返回已完成和未完成的任务。已完成的任务直接用task.result()获取未完成的统一取消并gather吞掉他们的取消异常。这样既保证整体不被打断也把所有分支都处理干净了。5.3 限速问题控制并发上限直接创建几十上百个create_task虽然没什么压力但目标服务可能扛不住或者对端有反爬限制。控制并发上限用的手段是信号量asyncio.Semaphore。信号量本质上是一个计数器。创建Semaphore(5)表示最多有 5 个协程同时进入受保护区域。超过的部分必须等前一个任务释放掉才能进来。LIMIT 5 sem asyncio.Semaphore(LIMIT) async def bounded_fetch(session, url): async with sem: # 只有 LIMIT 个协程能同时到这里 return await fetch_with_retry(session, url)把bounded_fetch当成普通协程去创建任务就能把并发数限制在 5 个以内。信号量在处理“异步任务的并发度控制”上是最常用的工具没有之一。我用它控制爬虫并发基本上没有把对方服务打爆过。5.4 把网络库从 requests 换成 httpx如果你的项目不想引入 aiohttp或者要同时兼容同步和异步代码我现在更多是httpx。这个库提供同步和异步两套 API代码风格几乎一致迁移成本很低。import httpx async def fetch_httpx(url): async with httpx.AsyncClient(timeout5) as client: resp await client.get(url) return resp.json()httpx在 Python 3.13 以后还被官方写进了标准库的 http.client 的后端演进计划里生态上会一直跟进。团队如果前后端浑然一体用httpx会减少在requests与aiohttp之间来回切换的割裂感。6. 常见问题与排查技巧实录这是我自己踩坑踩出来的一节记录下来省得你们再掉一次。每一个问题都来自实际运行中的报错或非预期行为排查思路也一并写清楚。6.1 TypeError: object asyncio.Future cant be used in await expression这个报错常出现在忘记await或把协程对象当成可等待对象时。常见场景是异步函数内部嵌套了另一种异步函数的调用但没有加awaitasync def main(): asyncio.sleep(1) # 忘了 await解决办法是检查所有调用协程函数的地方是否都加了await。另一种可能你把asyncio.Future当成普通函数调用或者把task.result()和await task混淆了。到这里我的建议是凡是协程对象先await再说。打印它的类型也能确认是不是协程对象。6.2 RuntimeError: no running event loop这个报错通常出现在“在非协程函数里尝试获取事件循环”时。比如直接在普通函数里调用asyncio.get_event_loop()或创建 Task。Python 3.10 以后没有运行中的事件循环时get_event_loop()在某些平台上会直接报错。正确的做法是如果你在一个协程内部用asyncio.get_running_loop()如果想从同步函数里启动异步流程用asyncio.run(coro)启动一个新的不要尝试从外部获取一个循环来“引导”它。6.3 Task exception was never retrieved这个警告很隐蔽也特别常见——任务抛了异常但没有任何地方去取它的结果。比如你create_task后就丢掉了引用没有await task也没有在gather里收集它。排查思路任务对象没有接收异常 异常被“吞”了会触发这个警告。最简单的方法是在创建任务后立刻绑定一个“处理异常”的回调async def task_that_might_fail(): raise ValueError(出错) async def main(): task asyncio.create_task(task_that_might_fail()) task.add_done_callback(handle_done) def handle_done(task): try: task.result() except Exception as e: print(f任务失败: {e})回调里调用task.result()会重抛异常这样我们才能接住。这个习惯我在写异步代码时是最先从没养成改为养成的——生产环境里很多“感觉没什么问题、偶尔漏数据”的 bug 其实都是任务异常被吞掉造成的。6.4 RuntimeError: Event loop is closed老版本 Python 里asyncio.run运行完后如果某些任务还在引用旧循环或者你在运行后尝试访问任务会报这个错。多数情况下是把loop.is_closed()误用了或者在asyncio.run之后还去调用 loop 相关的函数。排查建议不要把事件循环对象传出协程。如果你确实需要“后台任务在循环结束后继续”你要重新设计而不是想办法救活旧循环。异步的世界观里循环的生命周期就是应用生命周期循环结束一切归零。6.5 协程“不执行”新手最容易遇到的一个莫名其妙的现象写了async def调用了它结果函数体没有执行。原因其实在前面已经提了——你没有await它。任何协程对象只有被事件循环调度才会真正运行。你调用协程函数得到的只是一个“蓝图”蓝图不会自己盖楼。6.6 事件循环被阻塞如果有一个任务里做了time.sleep(5)或者做了很大的同步计算你会发现所有并发任务都像死掉了一样没有任何其他任务推进。排查的时候不要只看单个任务的代码要全局看有没有非await的阻塞调用。搜索time.sleep、requests、open().read()这类可能的阻塞点一个个改为asyncio.sleep、await asyncio.to_thread或await版本。6.7 问题速查表现象根本原因解决方案协程不执行没有 await 或没有放入事件循环用await或create_task调度类型错误无法 await对非可等待对象用了 await确认调用的是协程函数无事件循环运行在同步代码中获取循环用asyncio.run启动Task exception was never retrieved任务异常未被取用add_done_callback或统一gather事件循环被阻塞引入了同步阻塞调用替换为异步版本或使用to_thread事件循环已关闭循环结束后还访问旧循环不要跨循环保存任务引用7. 面向生产环境的落地建议最后分享一些我从实际项目中总结出的经验和习惯。这些都是文档里不太会写、但真正能用上的东西。7.1 为每个协程函数写日志上下文协程是并发的日志如果不带上任务标识多个协程交错输出的时候根本无法判断是哪条路径打的。我一般在 entry 处记录“任务开始”退出时记录“任务结束”和异常栈并且用 UUID 或 URL 作为关联 ID。import logging logger logging.getLogger(__name__) async def process_item(item_id): logger.info(开始处理 %s, item_id) try: await do_work(item_id) except Exception: logger.exception(处理失败: %s, item_id) else: logger.info(处理完成: %s, item_id)7.2 永远给外部依赖加超时网络请求不设超时最终一定会等到某一天某个下游系统半死不活整个事件循环卡住。凡是涉及aiohttp、httpx、aiomysql、redis等外部调用的地方全部设置超时。用一个公共的默认 timeout 常量避免每个函数都写死不同的值。7.3 写测试await 本身不保证顺序异步代码的功能测试要明确“协程并发执行顺序不受控”。断言结果时不要依赖“谁先谁后”。测试里对每个协程返回的内容做单元断言不要验证打印顺序。用pytest-asyncio插件可以让测试函数直接async def。7.4 注意 Python 版本差异asyncio从 Python 3.4 到 3.13 变化很大。我目前推荐的基线是 Python 3.11 以上。asyncio.timeout、asyncio.TaskGroup3.11 引入、asyncio.to_thread3.9 引入都让异步代码范式进步了不少。如果你还在维护 Python 3.8 的老项目很多新 API 用不了就得用ensure_future加wait_for的组合。特别是 3.11 引入的TaskGroup相比gather它更像一个作用域管理器任务组内任何一个任务失败组内其他任务的取消策略会更明确。新项目如果支持 3.11我建议优先考虑它。async def main(): async with asyncio.TaskGroup() as tg: task1 tg.create_task(fetch(a)) task2 tg.create_task(fetch(b)) task3 tg.create_task(fetch(c)) # 三个任务全部结束后继续往下走7.5 排查性能瓶颈的工具asyncio自带了一个调试模式PYTHONASYNCIODEBUG1或asyncio.run(main(), debugTrue)。开启之后它会检测“哪些协程执行时间太长”给出未 awaited 的协程警告以及阻塞事件循环的调用。另外loop.slow_callback_duration参数可以定义回调慢的阈值默认是 0.1 秒如果回调执行超过了这个时长就会打印警告。我从一个“主循环卡了 2 秒”的线上问题里拿到线索就是靠这个阈值警告最终定位到了某个第三方同步库在回调里做了网络请求。asyncio.run(main(), debugTrue)7.6 不要迷信“异步一定更快”最后泼一盆冷水异步并不天然比同步“快”。如果任务是 CPU 密集型的或者并发量本身很小同步代码逐行跑可能反而更简单、更快。异步的收益来自大量 I/O 等待被重叠。写代码前先想清楚瓶颈在哪儿不要让性能优化变成架构复杂化的借口。我一般会先用同步逻辑写出一个正确版本然后用cProfile或简单的耗时统计看一下瓶颈分布确认是 I/O 密集之后再改造成异步这样每一步都有对照组也不会在错误的抽象里耗掉大量时间。说实话Python 异步编程的门槛不在语法而在思维模式的转变从“从上往下一条路走到黑”变成“随时可以被打断、暂停、恢复”。一旦你理解了事件循环和任务调度的底层逻辑async、await和asyncio就是三个非常简单、配合默契的工具。我在日常工作里最喜欢的组合是用create_task自由创建任务用gather统一收集用wait做超时控制和选择性取消再用信号量拉高并发同时保护下游——这套组合拳打下来业务里 80% 的并发场景都能覆盖。先从一个小工具开始改造成异步跑一跑对比数据你会很快感受到这套机制的价值。