写了个脚本要批量请求几百个URL结果卡在原地干等网络响应CPU几乎没动静进度条半天走一格。相信不少朋友在Python里碰到过这种场景于是找到了async、await和asyncio这套异步编程方案。Python异步编程从3.4引入asyncio库、3.5正式提供async/await语法到现在已经是写高并发IO密集型任务的标配能力它让你在单线程内同时管理成百上千个网络连接、文件读写或数据库查询而不必为每个任务开线程、耗内存、拼切换。这篇实战总结会从“异步到底解决了什么问题”开始一步步拆解事件循环、协程、Task这几个核心概念然后用完整可复现的代码做一次并发请求与批量处理实战最后把我这些年踩过的坑、排查技巧和几个容易混淆的设计模式一并整理出来。适合刚接触异步的Python开发者也适合写过一些async代码但总觉得“差一口气”的朋友。1. 异步编程解决的核心问题别让CPU空等IO1.1 同步代码为什么慢等待全部是死等先看最传统的写法。假设用requests库逐个请求100个接口每个接口平均响应200毫秒整个流程大概需要20秒。问题不在于请求本身有多慢而在于“发出请求后等待响应的这段时间”程序什么都没干。同步请求的过程可以理解成你打电话给客服拨号之后一直握着听筒不说话直到对面接起来才继续交流。如果同时有10个电话要打你只能一个一个来每个电话的通话时间就是你的总时间。网络IO、磁盘IO、数据库查询这类操作有一个共同特点CPU发出指令后设备开始工作CPU就闲下来了但它还在“傻等”结果返回。一个进程里哪怕只有10个并发请求同步写法也会把它们变成10个串行等待单个请求的延迟被反复叠加。1.2 多线程方案的问题线程不是越多越好有人会说那用多线程啊每个线程处理一个请求不就行了早期我确实这么干过。concurrent.futures.ThreadPoolExecutor配合requests代码改动小效果也立竿见影。但随着并发量上来问题就暴露了每个线程有自己的栈空间动辄几十KB到几MB几千个线程直接吃掉大量内存。线程切换由操作系统调度线程越多上下文切换的CPU开销越大。Python的GIL全局解释器锁导致线程在CPU密集任务上无法并行IO密集任务虽然可以释放GIL但锁竞争依然存在。线程池大小需要反复调参调小了并发不够调大了反而更慢。我见过一个项目用ThreadPoolExecutor开200个线程去抢票结果服务端没挂本机CPU先飙到100%。这就是线程方案在超高并发下的典型困境。1.3 协程的本质在用户态自己安排“等待时间”异步协程的思路完全不同。它保留单线程单进程但是把“等待”变成“挂起”遇到IO等待时当前协程主动告诉事件循环“我先歇着你有别的活儿就干别的”等IO结果就绪了再回来继续执行。再拿打电话举例同步是挨个电话握着听筒死等协程是同时拨出100个电话拨通后才接听讲话。话务员只有一个人单线程但他不需要在每个电话上干等而是谁的声音来了就接谁的。这个过程涉及两个关键词async def定义的是一个协程函数调用它不会立即执行而是返回一个协程对象。await就是“挂起点”遇到它协程让出控制权事件循环去调度其他任务。真正管理这些协程的是asyncio的事件循环Event Loop。它是一个超级调度器维护着一个就绪队列和等待队列不停地在“检查IO状态、执行已就绪的协程、挂起等待中的协程”之间循环。1.4 什么时候该用异步IO密集是主场CPU密集别凑热闹用异步之前先做判断你的任务是IO密集还是CPU密集网络请求、文件读写、数据库操作、消息队列消费这类任务的特点是大部分时间花在等待上适合异步。图片处理、加解密、数据压缩、数值计算这类任务吃CPU协程帮不上忙反而因为调度开销更慢。这时候该用多进程或直接上numpy、Cython这类工具。实际操作中我判断标准很简单如果程序里大量时间都花在“等”上就用asyncio如果大量时间花在“算”上就别凑热闹。2. async/await与事件循环的工作原理2.1 协程不是线程也不是普通函数初学者最容易混淆的概念就是协程到底是什么从代码层面看它仍然是一个函数只不过用async def声明。但它的执行方式完全不同async def hello(): print(开始执行) await asyncio.sleep(1) print(执行结束)直接调用hello()不会打印任何内容只会返回一个coroutine对象。要让里面代码真正跑起来必须把它交给事件循环asyncio.run(hello())这个coroutine对象可以理解成一个“暂停的演员”它知道自己从哪开始、到哪暂停、暂停后从哪继续。函数里每个await都是一个暂停标记await右侧的表达式执行完毕之前协程不会继续往下走。协程最大的特征是“协作式调度”协程自己决定何时让出而不是被操作系统强占。这意味着协程只有在await处才会暂停普通计算代码一旦开跑就停不下来所以协程里千万不要写耗时很长的同步循环。2.2 事件循环核心调度器事件循环可以理解成一个“老板”手底下全是协程员工。员工们轮流汇报“我在等网络请求”“我在等数据库返回”老板就把他们挂到等待列表里然后去推进其他可以继续工作的员工。事件循环内部维护了多组数据结构_ready队列存放已就绪、可以立即执行的协程。_scheduled存放还没到时间的定时任务。各种IO事件监听器底层通过selectors模块监听文件描述符的可读可写状态。当await asyncio.sleep(1)执行时事件循环注册一个1秒后的定时回调然后把协程挂起。这1秒内循环去执行其他任务等时间到了再把协程放回就绪队列。asyncio.run()本质上是做三件事创建新的事件循环、把传入的协程作为第一个任务执行、最后关闭循环清理资源。它是Python 3.7引入的之前用loop asyncio.new_event_loop(); loop.run_until_complete(coro)现在统一推荐用asyncio.run。2.3 await到底在等什么可等待对象await后面只能跟“可等待对象”awaitable主要分三类协程对象coroutine来自async def函数的调用结果。asyncio.Task已经被事件循环调度的任务比裸协程多一层调度管理。asyncio.Future更底层的对象表示一个“将来才会有结果”的操作Task是Future的子类。一个关键区别直接await一个协程是“等它执行完”asyncio.create_task(coro)是把协程包装成任务、立刻丢给事件循环调度然后你可以在之后某个时间点await这个任务。前者是串行等待后者是并发安排。async def main(): task1 asyncio.create_task(hello()) task2 asyncio.create_task(hello()) await task1 await task2这个写法两个hello()才真正并发执行。如果写成async def main(): await hello() await hello()那就是严格串行第一个跑完第二个才开始。我见过不少半懂不懂的代码把async函数一个接一个await跑完发现性能毫无提升问题就出在这。2.4 Task的调度时机create_task只是把协程“注册”进事件循环不代表立即执行。事件循环会挑选合适的时机运行它。所以在create_task之后、await之前任务可能还没真正跑起来甚至在单任务场景下await就是给事件循环机会去执行它。这里必须理解await作为“挂起点”不只是等待结果更是“让出CPU、让事件循环有机会调度其他任务”的关键动作。没有await事件循环永远没有机会切换任务再多的create_task也不会并发。3. 完整实战并发请求、超时控制与并发限制3.1 场景设定先设定一个实际场景批量抓取100个商品详情页面的JSON数据解析出关键字段后写入本地文件。这个场景非常典型网络请求IO加文件写入IO全是异步的用武之地。基础环境说明Python版本3.10以上3.7都可运行但新版本对asyncio的API友好很多。需要安装aiohttppip install aiohttp。注意requests是同步库不能直接用在async函数里它的阻塞式IO会卡住整个事件循环。文件写入用aiofilespip install aiofiles。普通open().write()虽然语法不报错但它同步阻塞文件一大整个事件循环就僵住了。3.2 基础版用asyncio.gather并发执行先写一个最直接能跑通的基础版import asyncio import aiohttp import aiofiles import json async def fetch_one(session, url): async with session.get(url) as resp: resp.raise_for_status() return await resp.json() async def save_to_file(data, index): async with aiofiles.open(f./data_{index}.json, w, encodingutf-8) as f: await f.write(json.dumps(data, ensure_asciiFalse, indent2)) async def main(): urls [fhttps://api.example.com/product/{i} for i in range(1, 101)] async with aiohttp.ClientSession() as session: tasks [fetch_one(session, url) for url in urls] results await asyncio.gather(*tasks, return_exceptionsTrue) tasks [] for i, result in enumerate(results): if isinstance(result, Exception): continue tasks.append(save_to_file(result, i)) await asyncio.gather(*tasks) asyncio.run(main())几个关键点解释一下aiohttp.ClientSession建议全局复用不要每个请求新建一个。Session内部维护连接池反复创建销毁会浪费大量资源和时间。return_exceptionsTrue非常关键。gather默认遇到第一个异常就立刻抛出导致后面的任务被取消。生产环境里肯定不希望因为一个URL挂了就丢掉全部结果所以让异常作为返回值保留之后再逐个判断。这里的results是按传入tasks的顺序返回的不是按完成时间这个特性在需要“结果与请求对应”的场景特别省心。3.3 升级版加超时控制、并发限制与重试基础版能跑但离生产可用还差两样东西失控的并发量和慢请求拖垮整体。100个请求同时发出去本地可能承受得住但对端服务器未必。互联网上很多接口对单IP并发数有限制一口气打过去直接触发封禁。排查问题时要先把并发量压下来。使用Semaphore是最简单的限流手段import asyncio import aiohttp semaphore asyncio.Semaphore(10) # 最多同时10个请求 async def fetch_with_limit(session, url): async with semaphore: return await fetch_one(session, url)信号量的原理像商场限流门口保安数人头放进10个人里面人出来一个才放进下一个。这里的“人”就是协程“保安”就是信号量内部的计数器。超时控制同样不能少。一个API如果一直不返回任务就会一直挂在那占着资源。用asyncio.wait_for包一层给每个请求设一个最长时间async def fetch_with_timeout(session, url, timeout5): try: return await asyncio.wait_for(fetch_with_limit(session, url), timeouttimeout) except asyncio.TimeoutError: return Nonewait_for的原理是给内部任务设置一个超时回调超时就取消它并抛异常。取消动作本身也需要事件循环去执行不会一次到位。重试逻辑我习惯单独抽一个装饰器式函数尽量让主流程保持干净async def fetch_with_retry(session, url, retries3, timeout5): for attempt in range(retries): try: return await asyncio.wait_for(fetch_with_limit(session, url), timeouttimeout) except Exception as e: if attempt retries - 1: raise await asyncio.sleep(0.5 * (attempt 1)) # 退避注意重试时await asyncio.sleep(0.5 * (attempt 1))使用的是异步睡眠不会阻塞事件循环。新手很容易在这里顺手写time.sleep(0.5)协同程序一睡整个循环卡住其他99个请求全部等死。3.4 完整代码一个可复制的并发抓取脚本把限流、超时、重试、自动重命名文件、失败记录整合一下写成一个可以直接拿去改的完整脚本import asyncio import aiohttp import aiofiles import json from datetime import datetime CONCURRENCY 10 TIMEOUT 5 RETRIES 3 BASE_URL https://api.example.com/product/{} OUTPUT_DIR ./results FAILED_LOG ./failed.log semaphore asyncio.Semaphore(CONCURRENCY) async def fetch_one(session, url): async with semaphore: async with session.get(url) as resp: resp.raise_for_status() return await resp.json() async def fetch_with_retry(session, url): for attempt in range(RETRIES): try: return await asyncio.wait_for(fetch_one(session, url), timeoutTIMEOUT) except Exception: if attempt RETRIES - 1: raise await asyncio.sleep(0.5 * (attempt 1)) async def save_result(data, index): async with aiofiles.open(f{OUTPUT_DIR}/data_{index}.json, w, encodingutf-8) as f: await f.write(json.dumps(data, ensure_asciiFalse, indent2)) async def main(): urls [BASE_URL.format(i) for i in range(1, 101)] async with aiohttp.ClientSession() as session: tasks [fetch_with_retry(session, url) for url in urls] results await asyncio.gather(*tasks, return_exceptionsTrue) save_tasks [] failed_count 0 for i, result in enumerate(results): if isinstance(result, Exception): failed_count 1 continue save_tasks.append(save_result(result, i)) await asyncio.gather(*save_tasks) print(f完成成功 {len(results) - failed_count}/{len(results)}) if __name__ __main__: asyncio.run(main())这个脚本设计思路是限流压住对端压力超时防止单点拖累重试弥补临时故障异步文件写入避免IO阻塞。每个环节单独拆开都能复用组合起来就是一套可靠的小规模抓取骨架。3.5 同步与异步版本对比写完异步版本我特意用同步requests写了个等价对照用100个请求做实测对比模拟接口平均延迟200ms方案总耗时内存占用并发能力代码复杂度requests串行约20秒低1最低ThreadPoolExecutor(10线程)约2秒中10中asyncio aiohttp限流10约2.1秒低10中高asyncio aiohttp不限流约0.3秒低100中高结论很清楚协程方案在并发量大的场景下内存占用远低于线程方案而且不需要手动调整线程池参数并发上限由你写的信号量自行控制。如果对端服务完全不限流100个并发请求在异步下几乎瞬间完成。4. 事件循环的进阶操作与任务编排4.1 用asyncio.wait做到“部分完成即返回”gather的特性是所有任务都要等但有些场景只需要“最快的一个结果”。比如同时访问多个可用性相同的API谁先返回用谁的。此时asyncio.wait比gather更合适。async def fetch_fastest(session, urls): tasks [fetch_one(session, url) for url in urls] done, pending await asyncio.wait(tasks, return_whenasyncio.FIRST_COMPLETED) # 最快那个任务的结果 fastest_result done.pop().result() # 取消还没完成的任务 for task in pending: task.cancel() return fastest_resultreturn_when参数支持三种模式FIRST_COMPLETED第一个完成即返回。FIRST_EXCEPTION出现第一个异常即返回适合快速失败场景。ALL_COMPLETED全部完成才返回等价于gather默认行为。wait返回两个集合已完成的任务集合和未完成的任务集合。拿到done后记得把pending里的任务取消掉否则它们在后台继续执行白白消耗资源。4.2 用asyncio.Queue做生产者消费者有些场景任务不是一次性生成完毕的比如从分页接口持续拉数据、把数据分批写入数据库。这时候asyncio.Queue就派上用场了。import asyncio import aiohttp async def producer(session, queue): for page in range(1, 20): data await fetch_one(session, fhttps://api.example.com/list?page{page}) for item in data[items]: await queue.put(item) # 发送结束信号 await queue.put(None) async def consumer(session, queue): while True: item await queue.get() if item is None: break await process_item(item) queue.task_done() async def main(): queue asyncio.Queue(maxsize50) async with aiohttp.ClientSession() as session: producer_task asyncio.create_task(producer(session, queue)) consumer_task asyncio.create_task(consumer(session, queue)) await asyncio.gather(producer_task, consumer_task) asyncio.run(main())Queue在这里相当于一条流水线生产者从分页接口拿原始数据放到传送带上消费者从传送带上取数据逐条处理。maxsize50防止生产速度远超消费速度时内存暴涨。值得说明的是Queue的get和put都是异步的队列为空时get挂起等待队列满时put挂起等待。这种天然的反压机制让生产消费节奏自动匹配。4.3 协程之间的通信与结果传递协程之间不能直接共享变量应该通过返回值、队列、或者显式传参。初学者容易在协程函数内部修改一个全局字典然后发现结果并发写入乱套。正确的做法是在协程内部把结果返回由上层统一收集。也就是我前面示例的写法每个fetch_one返回自己的结果gather汇总成列表。如果协程之间必须传递数据用Queue或者asyncio.Event做同步。asyncio.Event适合一个协程等待另一个协程完成某个前置动作的场景。比如先登录拿到token再并发请求业务接口event asyncio.Event() token {} async def login_worker(): await asyncio.sleep(1) token[value] fake-token event.set() async def request_worker(): await event.wait() # 等登录完成 token token[value] # 继续请求4.4 多协程的优雅退出与超时兜底一个常见的线上问题是某个协程阻塞在第三方接口上无论怎么设wait_for都不退出整个进程无法正常关闭。这种情况通常是对端连接一直没有被关闭wait_for超时后触发了cancel但底层连接还在等待。兜底方案是给整个主流程加一个总超时try: await asyncio.wait_for(main(), timeout60) except asyncio.TimeoutError: print(主流程超时强制退出)另外事件循环关闭前应确保所有任务已完成或被取消。asyncio.run虽然会自动处理但如果你自己管理循环记得在退出前遍历所有任务def cleanup(loop): pending asyncio.all_tasks(loop) for task in pending: task.cancel() loop.run_until_complete(asyncio.gather(*pending, return_exceptionsTrue))5. 常见问题与调试技巧实录5.1 阻塞调用卡死事件循环time.sleep是头号杀手这是异步编程里最容易犯、后果也最严重的错误。在协程函数里使用time.sleep(1)代替asyncio.sleep(1)效果是整个事件循环冻结1秒所有并发任务全部停摆。为什么会这样因为time.sleep是同步阻塞调用它让当前线程休眠而事件循环在这个线程上运行自然跟着一起休眠。asyncio.sleep则是把“休眠”注册给事件循环协程挂起后让出控制权其他任务照常执行。同理在协程里调用requests.get()、open().read()、subprocess.run()都会阻塞事件循环。碰到这类需求要么换成异步库aiohttp、aiofiles、asyncio.create_subprocess_shell要么用loop.run_in_executor把同步操作丢到线程池执行。import asyncio async def call_blocking(): loop asyncio.get_running_loop() result await loop.run_in_executor(None, requests.get, https://api.example.com) return result5.2 为什么用了async却没有加速排查思路从代码结构入手。最常见的两种原因第一把所有async函数逐个await没有用create_task或gather并发编排写法变成了“异步的串行”。加速的前提是任务之间有独立的等待时间并且这些等待被并发了。第二某个协程内部执行的是CPU密集操作比如for循环里做了大量计算没有await点虽然函数声明了async但其他协程根本无法在它运行期间插进来。这就是前面说的“协程是协作式调度没有await就没有切换机会”。5.3 调试工具asyncio的调试模式与日志asyncio自带了调试模式开启后会在每个IO操作前记录耗时帮你在任务卡顿时找出“是谁占着资源不放”。在代码里设置import asyncio import sys if sys.flags.dev_mode: asyncio.run(main())或者运行脚本时加参数python -X dev script.py开启调试模式后事件循环会警告那些执行时间过长的回调打印出“Task was destroyed but it is pending”这类信息——后者通常是因为任务还没完成就被垃圾回收了或者忘记await任务。另一个实用技巧是给任务起名字排查并发问题时一目了然task asyncio.create_task(fetch_one(session, url), nameffetch-{i}) print(task.get_name())5.4 任务取消的坑CancelledError怎么处理task.cancel()会向任务内部抛入一个CancelledError如果协程内部不处理任务会静默终止。但如果在finally块里又执行了await就会抛出另一个CancelledError导致清理逻辑不完整。推荐的处理方式是使用asyncio.shield保护关键清理操作或者明确接收并吞掉取消异常async def worker(): try: while True: await do_work() except asyncio.CancelledError: # 做必要的清理 cleanup() raise # 重新抛出保持取消语义注意CancelledError在Python 3.8之后继承自BaseException而非Exception所以普通except Exception是捕获不到它的。很多人在协程里写了兜底异常处理但任务取消后却怎么都找不到原因问题就在这里。5.5 asyncio与多线程的边界有些项目试图在异步代码里使用threading.Lock或者在多线程环境里调用asyncio.get_event_loop()结果各种报错。几个明确的边界asyncio的Lock、Queue、Semaphore只能在协程里用不能跨线程直接用。在另一个线程中调度协程需要asyncio.run_coroutine_threadsafe(coro, loop)它会返回一个concurrent.futures.Future供线程间同步。多线程与多协程混合时建议明确区分职责线程负责阻塞式IO或CPU密集型任务协程负责高并发网络请求两者通过run_in_executor或run_coroutine_threadsafe作为桥梁。5.6 常用问题速查表现象可能原因解决方案协程代码“不执行”没有用asyncio.run或create_task调度确认协程被交给事件循环用async def但性能没提升逐个await导致串行用create_taskgather并发编排所有协程卡住不动协程内用了time.sleep、requests等阻塞调用换成异步库或run_in_executor任务报“was destroyed but pending”任务未await就结束生命周期保存任务引用确保await完成或cancelexcept Exception捕获不到取消CancelledError是BaseException单独写except asyncio.CancelledError某个请求超时不生效wait_for没用或有嵌套阻塞逻辑直接包住最终IO等待逐层检查对端服务器封禁IP并发量太高用Semaphore限制并发增加退避重试回调函数里用异步库报错回调是同步上下文不要在回调里await改用create_task包装6. 异步项目里的代码组织与设计建议6.1 按职责拆分IO层、业务层、调度层把异步代码写到一定规模后比如几千行就会觉得协程混在一起非常可怕。一个函数既发请求、又解析数据、又写文件、还管重试排查时根本分不清是哪个环节出问题。我习惯把代码拆成三层IO层只负责和外部系统打交道。fetch_one、save_result这一层参数是普通数据返回也是普通数据里面只做IO。业务层负责解析、清洗、校验数据不涉及任何IO。这一层甚至不需要async普通同步函数就行。调度层负责创建任务、控制并发、处理异常、编排流程。main函数和所有create_task、gather、Semaphore都在这一层。这样的好处是调换API、更换数据库驱动时只改IO层调整并发策略、限流规则时只改调度层业务逻辑变动完全不碰异步相关代码。6.2 别把所有函数都写成async一个常见误区项目里到处都是async def连加法运算都写成异步。这既难看又低效。实际上异步只有在“确实有IO等待”时才有价值。纯计算、纯字符串处理、纯数据转换写普通函数即可。如果业务函数内没有await却不小心把它写成了协程那么每次调用都要包一层Task调度开销反而更大。判断标准很简单函数内部是否出现await没有就不该加async。6.3 日志与监控异步代码的调试线索异步代码方便的地方在于单线程内极难看到“同时执行”的效果也因此排查并发问题时如果没有日志完全两眼一抹黑。我通常在关键节点打日志记录任务开始、完成、失败和耗时import time import asyncio async def log_wrapper(coro, name): start time.perf_counter() try: result await coro print(f[{name}] 完成耗时 {time.perf_counter() - start:.2f}s) return result except Exception as e: print(f[{name}] 失败{e}) raise批量任务里耗时明显偏长的名字往往是网络抖动或接口慢请求的线索。日志里附上任务名能快速定位到具体请求。6.4 版本兼容性与API变迁asyncio的API变化挺大老项目升级Python版本时容易踩坑Python 3.7之前asyncio.get_event_loop()在无当前循环时会创建并设置新循环3.10之后在协程内部调用它会抛RuntimeError。asyncio.run是3.7引入之前版本只能手动管理循环。asyncio.current_task替代了老的asyncio.Task.current_task()。Python 3.10之后asyncio.get_event_loop()的默认行为是如果当前没有正在运行的循环抛出DeprecationWarning乃至RuntimeError。3.11后协程任务异常处理、TaskGroup也做了增强asyncio.TaskGroup可以自动管理一组任务的错误传播和取消。写跨版本兼容代码时优先使用asyncio.run、asyncio.create_task、asyncio.current_task这些“新而稳”的API避免直接操作底层loop。6.5 与第三方异步库配合时的注意事项异步生态里每个第三方库都要有对应的异步实现绝不能混用同步版本。常见的配套关系HTTPaiohttp对应requests。文件aiofiles对应open()。数据库asyncpg、aiomysql对应psycopg2、pymysql。Redisredis.asyncio对应redis。ORMSQLAlchemy异步模式或Tortoise ORM。选型时先看库是否实现了异步协议没有就找替代。混用同步库到协程代码里事件循环照样被阻塞。另一个注意事项是某些同步库内部实现C扩展并持有GIL即使用run_in_executor也不一定能完全并行需要重点关注。7. 参考模式与个人实操心得7.1 一个典型的生产消费编排模板很多IO密集型系统都可以抽象成“获取数据、处理数据、落盘/入库”三段式。我总结出一个比较通用的模板新项目起步时可以直接套用async def run_pipeline(concurrency20): semaphore asyncio.Semaphore(concurrency) queue asyncio.Queue(maxsize100) async def producer(): # 获取数据源 async for item in fetch_all_items(): await queue.put(item) await queue.put(None) # 结束标记 async def worker(): while True: item await queue.get() if item is None: break async with semaphore: await process_one(item) queue.task_done() async with aiohttp.ClientSession() as session: # 每个worker实际上可以共用同一个session workers [asyncio.create_task(worker()) for _ in range(concurrency)] await producer() await asyncio.gather(*workers)这个模板把获取、分发、执行、收尾都拆开并发数通过concurrency一个参数控制够用且清晰。7.2 什么时候坚决别用异步异步并非银弹有几类场景我倾向直接拒绝异步方案项目规模很小总共三五个请求串行也能接受没必要引入异步框架增加理解成本和心智负担。团队中没人懂事件循环后续维护大概率出问题写出来的“异步代码”可能比同步代码更慢、更乱。对端服务完全同步且无法配合比如只能串行交互的全双工协议异步带来的收益有限。任务本身是CPU密集计算且没有调用外部服务这时候异步约等于原地打转。7.3 从同步到异步我的迁移经验如果手头有一套运行良好的同步代码不要一次性全部重写成异步。我推荐步步为营的迁移路径第一步先理清IO边界哪些地方在等待外部系统哪些在做计算标注出来。第二步把耗时最长的IO操作改造成异步单独跑一个异步入口验证事件循环能正常工作。第三步逐步把其他同步IO替换为异步库用Semaphore控制并发压测对比指标。第四步清理平台化的问题定时任务调度、信号处理、日志刷新等。整个迁移过程要持续跑同一批回归测试确保逻辑一致。7.4 最后再分享一个排查卡死问题的方法遇到协程“卡死”的疑难杂症直接可以用faulthandler打印当前执行栈一眼就能定位卡在哪一行python -X faulthandler script.py或者代码里主动开启import faulthandler import sys faulthandler.enable(sys.stderr)这套组合拳我救过几次场上次生产环境一个采集任务突然卡住就是靠它定位到有段代码在协程里偷偷用了time.sleep(10)三个worker全被卡死整个队列堵了五分钟。从那以后我给自己定了一条规矩协程函数里禁止出现任何同步阻塞调用审查代码时看到time.sleep、requests.get、open().read()直接打回。这套经验帮助团队少踩了无数坑希望也能帮到你。