你一定遇到过这种场面项目里新的接口都是async def但老的同步代码里就是调不动它。直接写result fetch()只会拿到一个 coroutine object控制台还会阴阳怪气地冒出一句 “coroutine was never awaited”让你反复怀疑自己是不是忘了加await。我当时被这个问题卡住的场景是一个 Django 定时任务脚本里要调用内部已经全部改成异步的服务脚本本身却是一个普通的同步函数直接调异步函数根本拿不到结果。“异步协程转同步执行”说的就是这件事把本应该用await等待的协程包装成一个同步调用的接口让调用方看起来就像在调用普通函数一样拿返回值、处理异常一概不变。这个需求常见于测试用例、老系统迁移、混合架构的粘合层以及所有“调用方还没准备好接触异步”的边界场景。接下来我会把背后的原理、几种主流语言的转换姿势、还有我在实战里踩过的坑一次性讲清楚。1. 为什么会有“异步协程转同步执行”这个需求1.1 异步与同步的本质差异同步执行是“一步步来”。函数 A 调用函数 B如果 B 内部要等一个网络请求当前线程就停在那里直到 B 返回A 才能继续往下走。这种方式的好处是逻辑直观坏处是等待期间线程资源被白白占用。一个经典例子是连续请求两个 HTTP 接口同步写法是一个一个来假设每个耗时 1 秒两个就是 2 秒。异步执行则是“先挂起等通知”。同样是两个 HTTP 接口异步写法可以同时发起请求谁先返回就先处理谁。Python 里的asyncio让业务代码用async def声明协程函数用await标记真正的挂起点。执行到await时当前协程会主动让出控制权事件循环立刻去调度别的任务等这个 await 对应的操作完成后再回来继续执行。这样两个请求的总耗时约等于最慢的那个而不是简单的相加。我用一个生活化的方式理解同步像是食堂打饭一个窗口一个窗口排你必须在这队里等完才能去下一队异步像一个有经验的服务员先把好几桌客人的单都下了然后哪桌菜好了就先端哪桌。异步并不一定“更快”但它在等待 IO 的场景下能大幅提升吞吐量。不过代价是所有异步代码都被传染一旦某个函数用了async def调用它的人也需要进入异步上下文这就是标题里“转同步”需求的根本来源。1.2 哪些场景真的需要“转同步”我这几年的实际经历里最常碰到“异步协程转同步执行”需求的有这么几类老系统里调用新异步服务。比如旧项目还是 Flask/Django 的同步视图但内部服务已经切到httpx.AsyncClient、aiohttp这类异步客户端直接调会拿不到结果。测试框架不支持异步。很多测试框架的默认执行模型是纯同步的要把pytest跑出异步步骤要么装pytest-asyncio要么写一个同步函数把异步协程转同步执行。命令行脚本和定时任务。它们是程序的顶层入口天然适合同步调用你可以直接用asyncio.run()跑完整个 async 主函数。消息队列消费者、周期任务。Celery、APScheduler 等任务调度器默认调用同步函数想在里面调用异步服务就需要一个同步壳。给外部系统提供 RPC/HTTP 接口。调用方不关心你是不是异步只关心能不能拿到结果。这时需要在接口边界做一次异步转同步。1.3 转换前必须想清楚三个问题“异步转同步”听起来简单但动手前必须先回答三个问题否则后面全是坑第一阻塞当前线程是否可接受所谓“转同步”本质上是让当前线程停下来等协程跑完。如果你在一个线程池很紧张的系统里大量使用这种转换线程会被占住系统的并发能力会直线下降。第二事件循环的生命周期由谁负责Python 里asyncio.run()每次会创建新的事件循环用完就关闭而某些运行环境里已经存在一个事件循环正在运行再用asyncio.run()就会直接报RuntimeError: asyncio.run() cannot be called from a running event loop。第三这段转换代码会不会被再次放进异步环境如果你写了一个同步函数内部做异步转同步结果这个同步函数又被另一个异步函数调用那线程被阻塞的同时外层事件循环也停了。这是很多死锁和性能问题的伏笔。2. 核心原理协程转同步的底层发生了什么2.1 协程对象和任务对象的区别在 Python 里async def定义的函数被调用时并不会立即执行函数体而是返回一个coroutine object。这个对象就像一份“待执行的清单”它记录了函数从哪里开始、中间有哪些await断点、目前为止的状态是什么。你不把它交给事件循环它永远是凉的。要让这个 coroutine 真正跑起来通常要把它包装成Task。Task 是事件循环里的一个调度单元它负责跟踪协程的执行进度并在协程被await挂起时将控制权交还给事件循环。简单理解协程是状态机Task 是运行在这个状态机上的调度器外壳。这个设计不是 Python 独有的。Kotlin 协程同样把启动协程和调度分开launch或async创建的协程需要协程作用域CoroutineScope来承载JavaScript 的 async 函数本质上也是协程只是它的调度被内建在引擎的 event loop 里。2.2 事件循环是怎么跑起来的事件循环是一个巨大的while循环。它维护着一堆队列和数据结构就绪队列包含所有当前可以继续执行的 Task。延迟队列包含那些需要等待一段时间才唤醒的 Task比如await asyncio.sleep(1)。IO 监听器当某个 socket 可读/可写时唤醒对应的 Task。循环每转一圈会做三件事检查是否有到期的延迟任务把它移到就绪队列检查 IO 事件把等待该事件的 Task 唤醒从就绪队列里按顺序取出一个 Task让它执行到下一个await为止。这就是为什么异步代码能在单线程里看起来“并发”。如果你要跑loop.run_until_complete(coro)事件循环会一直执行直到传入的这个 coroutine 对应的 Future 完成才停止循环并返回结果。所以“转同步”并不是魔法它只是把“当前线程”直接卡在事件循环的运行上等目标任务结束。2.3 “转同步”的代价到底是哪里堵住了很多人以为asyncio.run(coro)是“直接把协程变成同步函数”这是误解。它的完整流程是新建一个事件循环 → 把协程打包成 Task → 运行事件循环直到该 Task 完成 → 关闭事件循环。在这个过程中事件循环确实在运行所以它不仅执行你传入的那个协程也可能顺手执行了其他就绪任务。但调用asyncio.run()的当前线程是实实在在阻塞住的——函数不会返回直到协程完成。这意味着如果你是在一个已有的异步上下文里调用它外层事件循环会被这个同步调用卡住只要内部协程中有任何阻塞操作整个程序就停转了。这个代价必须在做架构决策时就清楚异步转同步适合放在程序的最外层边界适合作为一次性入口不适合放在业务热点上反复横跳。3. Python 实操把协程跑成同步的几种姿势3.1 顶层入口优先选 asyncio.run()如果你的同步上下文是整个程序的入口比如命令行脚本、单元测试函数、定时任务的入口函数那asyncio.run()是最简单直接的做法。import asyncio async def fetch_user(user_id: int) - dict: # 这里可能是请求远程 API 或查询数据库 await asyncio.sleep(1) return {id: user_id, name: zhang} def main(): # 普通同步函数直接返回协程结果 user asyncio.run(fetch_user(42)) print(user) if __name__ __main__: main()asyncio.run()会自动帮你创建新事件循环、执行协程、关闭事件循环。它是 Python 3.7 以后官方推荐的顶层入口写法代码最干净也不会出现“上次的 loop 没关闭”这类遗留问题。不过要注意asyncio.run()不能在已有运行中的事件循环里调用也不能在一个已经关闭的 loop 上重复调用。它每次都会创建一个全新的 loop所以如果你在进程里调用它很多次每次都会产生创建和销毁的开销。虽然这个开销通常不大但如果在高频热路径上反复调用性能会很难看。3.2 处理已存在事件循环的场景run_until_complete如果你发现自己已经处于一个有事件循环的环境里比如 Jupyter Notebook、某些 Web 框架的 shell或者想在一个进程里复用同一个 loop就要避免asyncio.run()。这时可以用loop.run_until_complete()import asyncio async def do_work(): await asyncio.sleep(1) return done loop asyncio.new_event_loop() try: result loop.run_until_complete(do_work()) print(result) finally: loop.close()这里我用了asyncio.new_event_loop()手动创建新循环避免触碰到环境里已经运行的那个 loop。核心思路是既然当前环境的事件循环不能碰那就隔离开在新线程里创建一套独立的事件循环。如果你是在异步函数内部被动地需要同步等待一个协程结果问题会更棘手。一个常见做法是把转换代码扔到线程池里执行让外层异步事件循环继续运行import asyncio from concurrent.futures import ThreadPoolExecutor async def main(): loop asyncio.get_running_loop() result await loop.run_in_executor(None, sync_run_coroutine, do_work()) print(result) def sync_run_coroutine(coro): inner_loop asyncio.new_event_loop() try: return inner_loop.run_until_complete(coro) finally: inner_loop.close()这种方式能绕开“事件循环已存在”的报错因为新的协程是在独立线程的独立循环里执行的。代价是线程切换、资源隔离、上下文传递都要自己管理好别在里面塞不可复用的资源。3.3 Django/Flask 老项目里用 async_to_sync在 Web 老项目里最常见的是 Django 同步视图调用异步函数。Django 自带的asgiref.sync提供了一个非常顺手的工具async_to_sync它专门用于把异步函数包成同步函数。from asgiref.sync import async_to_sync import asyncio async def get_user_details(user_id: int) - dict: await asyncio.sleep(1) return {id: user_id, status: active} # 转成同步函数 sync_get_user_details async_to_sync(get_user_details) def my_django_view(request): user_id request.GET.get(user_id, 1) details sync_get_user_details(user_id) return JsonResponse(details)这里最直观的感受是你不用自己管理事件循环、不用手动去处理“当前是不是已有 loop”的报错async_to_sync内部已经把线程和事件循环的切换处理好了。它还会做一层asyncio.run和线程的适配即使调用方处于异步环境它也能通过开新线程的方式避免死锁。我用这个方案做过几个老项目的接口升级体验是它适合做薄适配层不适合在业务代码里到处包。如果你发现一个视图里出现十几个async_to_sync包出来的函数说明数据结构本身应该重新设计而不是继续堆适配层。3.4 别在异步内部“假装同步”另一个容易踩坑的点是协程里不应该用time.sleep或同步阻塞库来“等一件事完成”。常见错误# 错误示范会阻塞整个事件循环 import time async def bad_example(): time.sleep(2) # 这会让所有协程都停下来 return bad正确做法是用await asyncio.sleep(2)或者如果需要执行一段同步阻塞 CPU 任务可以用asyncio.to_thread把它扔到线程池import asyncio import time def heavy_sync_calc(n: int) - int: # 模拟耗时的同步计算 time.sleep(2) return n * 2 async def main(): result await asyncio.to_thread(heavy_sync_calc, 21) print(result) # 42这虽然不是“异步转同步”但是“同步函数转异步执行”两者经常成对出现。当你把异步协程转同步调用后如果内部又包含同步阻塞代码最终效果会回到同步执行的模型上性能优势荡然无存。所以转换的时候要一并检查协程内部有没有不该出现的同步阻塞调用。4. Kotlin 与其他语言的对照方案4.1 Kotlin 的 runBlocking一种真正的阻塞Kotlin 协程生态里最典型“异步协程转同步执行”的工具是runBlockingimport kotlinx.coroutines.runBlocking import kotlinx.coroutines.delay suspend fun fetchUser(): String { delay(1000) return user } fun main() { val user runBlocking { fetchUser() } println(user) }runBlocking会创建一个新的协程作用域并阻塞当前线程直到作用域内所有协程执行完毕。这个名字取得非常直白它就是用一个阻塞动作把 suspend 函数包成了同步调用。但 Kotlin 社区对它的态度也很有趣runBlocking主要用于main函数、测试和桥接旧代码如果把它用在 UI 线程或者高并发的控制器里很容易造成界面卡顿或线程饥饿。Android 开发者应该深有体会runBlocking跑在 UI 线程上就是灾难因为 UI 线程被阻塞后连绘制都没法执行。本质上runBlocking和 Python 的asyncio.run()在“阻塞当前线程等待结果”这一点上是一致的。区别在于 Kotlin 的协程调度器更丰富runBlocking默认用事件循环式的调度器内部遇到delay也不会死等而是把线程空出来执行其他协程。但一旦你从外部看这个调用仍然是阻塞的。4.2 为什么 JavaScript 里几乎没有“转同步”这回事JavaScript 的 async/await 也被某些人称作协程但它和 Python、Kotlin 有一个很大的区别JS 的事件循环是单线程且不可阻塞的。如果你在浏览器里写一个同步等待、阻塞直到 Promise 完成整个页面会直接假死在 Node.js 里如果你用某些同步代码去阻塞事件循环服务的并发能力会瞬间归零。所以 JS/TS 的选择通常是要么把整个调用链改成 async/await要么在模块顶层使用顶层 await。Node.js 生态里没有官方提供的 “runBlocking” 或 “asyncio.run”因为这种操作从根本上违背了单线程事件循环的设计原则。理解了这一点你对 Python 中“异步转同步”的价值也会有更深的认识它之所以可行是因为 CPython 的 asyncio 有自己的事件循环实现并且在阻塞时还能把控制权交给操作系统调度但它依然要付出线程被占用的代价。5. 常见问题与排查技巧实录5.1 RuntimeError事件循环已经存在我在给一个老脚本接入异步时最常遇到的是这个报错RuntimeError: asyncio.run() cannot be called from a running event loop原因很直接当前线程已经有一个事件循环在运行asyncio.run()不允许再创建新的。这种情况在 Jupyter、IPython、某些测试框架里特别常见因为这些环境本身已经启动了一个 loop。我的排查思路分两步先确认当前代码是不是已经在一个协程里。如果是就不要试图在内部调用asyncio.run()而是应该让外层一起写成 async/await或者用asyncio.run_coroutine_threadsafe把协程调度到另一个线程的 loop如果确实需要在同步上下文里调优先用asyncio.new_event_loop()在独立线程里跑避免和当前 loop 冲突。5.2 coroutine was never awaited 的灵异事件新手最容易卡住的是“函数调用了但结果不对”。比如async def fetch(): return hello print(fetch()) # 输出 coroutine object不是 hello这很像是程序抽风实际上它完全是语言设计。fetch()只是创建了协程对象并没有执行。要让协程运行必须把它交给事件循环或者用await等待。排查时先检查是不是少了await在同步代码里是不是没有用asyncio.run()包一层。这类报错往往还伴随着令人困惑的警告sys:1: RuntimeWarning: coroutine fetch was never awaited这时候别急着删代码去调用链的上游看谁调用了这个函数它是否真的处于 async 上下文。5.3 反复调用 asyncio.run() 导致资源被关闭asyncio.run()每次都会创建并关闭一个事件循环。如果你在循环里频繁调用它每次都会销毁旧的 loop这会导致一些“依附于 loop”的资源被提前清理。比如你在协程里创建了一个 asyncio 的锁、队列或者调用了asyncio.get_event_loop()缓存下来的东西可能在第二次asyncio.run时就已经失效了。我见过有人在一个循环里 1000 次调用asyncio.run()去处理独立任务结果每隔一段时间就出现连接被关闭、锁状态丢失的诡异问题。后来把方案改成了“外部创建一个长期 live 的 loop内部用run_until_complete反复执行”问题就消失了。所以如果你的进程要多次执行异步任务应该考虑复用事件循环而不是每次都从零启动。5.4 阻塞 IO 放错位置把整个事件循环卡死另一个高发问题异步转同步后业务运行正常但系统的整体吞吐量反而下降了。典型原因是协程内部用了同步阻塞库比如requests.get()、time.sleep()导致事件循环在等待期间无法调度其他任务。排查方法很直接看协程里有没有同步网络库、同步文件读写、同步数据库驱动。如果有一是换成对应的异步版本比如用httpx.AsyncClient替代requests二是实在换不了用asyncio.to_thread把同步阻塞部分挪到线程池。只有把阻塞点移出事件循环异步转同步才有意义。5.5 常见问题速查表现象可能原因解决方向asyncio.run() cannot be called from a running event loop已经在事件循环里调用 asyncio.run改用asyncio.run_coroutine_threadsafe或新线程 new loopcoroutine was never awaited没把协程对象交给事件循环补await或外层用 asyncio.run 包住程序反复出现奇怪资源失效多次 asyncio.run 导致 loop 重建复用一个 loop用 run_until_complete进程慢、吞吐量下降协程内部有同步阻塞调用换异步库或用 asyncio.to_threadUI/服务线程卡死在高频路径上做同步阻塞等待把同步适配放到最外层入口避免内层反复转换6. 适配层设计把“转同步”做成一件干净的事6.1 设计一个能复用的 SyncRunner既然天天要转换不如做一个通用的“同步执行器”把事件循环的创建、执行、关闭都封装进去。这样业务代码里不用反复写asyncio.new_event_loop()这种细节。import asyncio from concurrent.futures import ThreadPoolExecutor class SyncRunner: def __init__(self): self._loop asyncio.new_event_loop() def run(self, coro): if asyncio.get_event_loop().is_running(): # 当前环境已有事件循环丢到另外的线程里跑 with ThreadPoolExecutor(max_workers1) as executor: return executor.submit(self._run_internal, coro).result() return self._run_internal(coro) def _run_internal(self, coro): return self._loop.run_until_complete(coro) def close(self): self._loop.close()这个思路的核心是同步转换不是散落在代码里的补丁而是一个有明确边界的适配器。谁要调用异步服务谁就去依赖这个SyncRunner而不是自己在模块里到处 new loop、销毁 loop。长期维护起来至少不会出现“每个文件里都有asyncio.new_event_loop()”这种失控局面。6.2 让同步接口和异步业务分离如果你正在做新项目我强烈建议把“同步入口”和“异步业务”分成两层。同步层只负责接收请求、调用适配器、返回结果异步层只写async def不关心调用方是同步还是异步。# 业务层永远是 async def async def order_service(user_id: int): user await fetch_user(user_id) cart await fetch_cart(user_id) return await create_order(user, cart) # 同步入口层负责转换 def create_order_sync(user_id: int): return sync_runner.run(order_service(user_id))这样做的好处是将来如果业务层被一个纯异步的服务接入可以直接用内部的async def不需要把已经转同步的代码再绕一圈拆回去。很多老项目最痛苦的“异步化改造”本质上就是当初没有把同步边界留出来结果所有函数都互相纠缠无法精确切割。6.3 别忘了超时、取消和日志异步协程转同步后很多开发者会丢掉异步时代非常有用的能力超时、取消、上下文日志。比如在异步里你可以用asyncio.wait_for(coro, timeout2)给任务加超时但转成同步函数后如果不加任何处理一旦背后的 IO 挂了同步调用就会无限期阻塞。解决方法是在做适配层时把超时参数一并暴露出来。async def fetch_with_timeout(coro, timeout: float): return await asyncio.wait_for(coro, timeouttimeout) def fetch_sync(user_id: int, timeout: float 3.0): return sync_runner.run(fetch_with_timeout(get_user_task(user_id), timeout))这样同步调用方至少能拿到超时异常而不是干等。日志也是同理转同步的适配层应该记录从哪个同步入口进入、对应哪个协程任务、耗时多少否则排查线上问题的时候日志会缺失一大块上下文。最后再分享一个小技巧我现在的原则是“只在最外层转一次”。脚本的main函数、测试用例、Web 框架的适配器这些地方可以大大方方用asyncio.run、async_to_sync或runBlocking来做异步协程转同步执行但业务代码内部如果频繁出现同步等待我就会停下来想一想是不是设计出了问题。把阻塞边界守住把事件循环的生命周期管好异步和同步才能在同一套系统里和平共处。