FastAPI异步方法中调用同步方法你们是不是也这么踩过坑前几天调一个FastAPI接口在异步方法里直接调用了一个第三方SDK的同步方法单测全绿一发到测试环境压测就发现请求延迟越堆越高。最后定位到问题async def里面同步调用把事件循环堵住了。FastAPI异步方法里调同步方法这个坑几乎每个写FastAPI的人都会遇到而且它不像编译报错那么显眼更多时候是“看起来没毛病一压测就露馅”。这篇文章就把这个问题的原理、报错、正确写法、实战改造一次讲透适合所有用Python FastAPI写接口的同学参考。1. 先搞清楚FastAPI的同步与异步到底是谁在调度谁1.1 async def和defFastAPI是两套处理逻辑很多人刚开始写FastAPI时会有一个误解只要接口函数用async def定义就能“并发处理”请求用def定义就只能串行处理。这个理解不完全对。FastAPI底层根据inspect.iscoroutinefunction判断函数类型如果接口是async defFastAPI会把协程交给asyncio事件循环去调度如果接口是普通defFastAPI不会直接调用它而是把这个同步函数丢进线程池再在线程池里执行。线程池的作用很关键。默认情况下FastAPI使用anyio的线程池来执行普通def接口所以即便你所有接口都写成普通defFastAPI一样能并发处理很多请求只是这里的并发靠的是多线程而不是协程。真正的问题出在混合使用场景高并发核心接口用async def保持协程调度效率但函数体里又忍不住直接调用同步库比如requests、time.sleep、同步数据库驱动、本地文件读写、加密解密等。这一下就把事件循环卡死了。路由写法执行方式是否阻塞事件循环适用场景def丢入线程池执行否在线程池中阻塞同步IO、简单计算、不涉及其他异步任务async def事件循环直接执行会若函数内有同步阻塞调用异步IO、高并发、需要同时等待多个IOasync def 线程池封装同步调用协程挂起同步调用在线程池执行否在异步接口中调用同步库的最佳实践FastAPI官方文档也建议如果你不确定某个函数是放async def还是def在没有异步依赖时直接写def更省心。但现实项目的复杂度在于路由层已经是async def业务方法又大量使用同步方式引入这时候就只能靠开发者自己把同步调用包起来。1.2 事件循环是单线程的阻塞等于所有请求排队认识事件循环可以用一个生活场景类比把事件循环想象成一个只有一个收银员的奶茶店。async def接口正常情况下是“点单后拿号去做别的事等提示好了再回来取”。但如果某个async def里直接调用了一个同步方法相当于收银员在给你做奶茶时突然原地愣住10秒不接待下一个人这时所有排队的人都被卡住。asyncio的事件循环是单线程模型同一时刻只能执行一个任务。它靠协程主动挂起(await)来切换任务而不是靠操作系统的抢占式调度。同步方法一旦在事件循环线程里启动loop就会一直等它返回期间无法处理任何网络请求、无法处理定时任务、无法调度其他协程。这就是为什么在async def里裸调同步方法会“一个接口拖垮整个服务”。理解这一点后很多现象就有了解释为什么压测时并发越高延迟反而疯狂上涨为什么只是调一个慢速第三方接口却让你的健康检查接口都开始超时。本质上都是同一个问题有人在事件循环里干了不该干的同步活。2. 在async def里调用同步方法为什么说它是灾难2.1 阻塞事件循环并发能力直接掉底写个最简单的例子from fastapi import FastAPI import time app FastAPI() app.get(/sync_in_async) async def bad_case(): time.sleep(2) # 模拟同步阻塞比如requests.get、文件读取 return {msg: ok}这个接口虽然用了async def但函数体内没有任何await挂起点只做了一次2秒的同步sleep。当50个请求同时打到这个接口时事件循环里这50个协程排队执行每个都要在事件循环里阻塞2秒最后一个请求的响应时间至少是100秒。更可怕的是服务器上所有其他async def接口也会被拖慢因为它们共用同一个事件循环。把同样的50个请求改到下面这个版本from fastapi import FastAPI import time from starlette.concurrency import run_in_threadpool app FastAPI() app.get(/good_case) async def good_case(): await run_in_threadpool(time.sleep, 2) return {msg: ok}同步的2秒消耗从事件循环转移到了线程池整体并发能力立刻恢复。这里要说明一点单次请求的响应时间并不会变快依然是2秒多但系统能同时处理的请求数量明显提升不会因为一个慢接口把所有接口拖垮。2.2 “await运算符只能用于异步方法中”这个报错到底是怎么回事搜“FastAPI异步方法调用同步方法”时经常能看到这句报错信息“错误1‘await’运算符只能用于异步方法中。请考虑用‘async’修饰符标记此方法”。这句报错专门说给C#等编程语言听的Python的报错信息通常是SyntaxError: await outside async function或者TypeError: object int cant be used in await expression。不管哪个语言本质都一样await只能等待“可等待对象”。可等待对象包括协程、Future、Task。普通同步方法的返回值是常规对象不是可等待对象所以你写await sync_func()根本等不了它。常见误用有两种在普通def函数里写了await some_coroutine()Python直接报语法错误因为当前函数不是async。在async def函数里对一个同步函数写了await sync_func()解释器会报“xxx对象无法用于await表达式”。我见过不少同学把这两种错误混在一起以为把函数加上async修饰符、然后把所有方法调用都加上await就万事大吉。这恰恰把问题搞复杂了。在异步方法中调用同步方法关键不是给同步方法加不加async而是怎么让同步代码离开事件循环。如果同步函数本身没有IO等待、只是几行CPU计算你把它改成async也没有意义如果同步函数是一个阻塞调用直接await更是错的因为同步函数返回的不是协程对象它不会在await时暂停等待而是直接执行完才返回。2.3 同步调用带来的三类隐藏成本阻塞事件循环是最直接的问题但同步调用在异步代码里的成本还有另外两类。第一类是线程切换成本。如果用run_in_threadpool把同步方法丢进线程池每个调用都会涉及一次协程挂起和恢复、线程池任务调度这部分开销比纯异步IO略高。所以在可以用异步库替代同步库的场景下优先换库线程池只是兜底方案。第二类是资源占用成本。大量同步调用涌进线程池时线程数量会瞬间膨胀。每个线程都有自己的栈空间如果某个同步方法内部还申请了锁、数据库连接、文件句柄线程多了之后资源开销会很吓人。这就是为什么排查问题时需要先看看是不是线程池被占满再决定要不要扩大线程池。第三类是并发安全成本。同步方法一旦被丢进线程池它就和事件循环里的协程并行运行了如果有共享可变状态就需要加锁或改用线程安全的数据结构。这一点新手很容易忽略以为“反正都是在FastAPI框架里应该不会有线程安全问题”。3. 正确的调用姿势把同步方法请出事件循环3.1 方案一用starlette自带的run_in_threadpoolFastAPI基于Starlette而Starlette自带一个非常好用的工具函数run_in_threadpool。它内部封装了anyio.to_thread.run_sync会把同步函数提交到事件循环关联的线程池去执行再以协程方式等待结果。from fastapi import FastAPI from starlette.concurrency import run_in_threadpool app FastAPI() def sync_query(user_id: int): # 模拟一个阻塞操作比如同步ORM查询、requests请求 return {user_id: user_id} app.get(/users/{user_id}) async def get_user(user_id: int): # 关键用await包裹run_in_threadpool result await run_in_threadpool(sync_query, user_id) return result注意参数传递方式run_in_threadpool的第一个参数是要执行的同步函数后面的参数会按位置传给同步函数。如果同步函数需要关键字参数可以用functools.partial先绑定或者写一个lambda包一层。run_in_threadpool同样也支持关键字参数直接传入在较新版本中但为了兼容性我习惯用functools.partial。这个方案适合所有IO密集型的同步调用同步数据库查询、同步HTTP请求、同步文件读写、同步邮件发送等。它的最大优点是零依赖FastAPI项目开箱即用。3.2 方案二用anyio.to_thread.run_sync更加底层通用FastAPI底层依赖AnyIO所以anyio.to_thread.run_sync在任何FastAPI项目里也是直接可用的不要求你额外安装包。它比run_in_threadpool更接近底层实现同时支持asyncio和trio两个后端在控制协程取消行为时更灵活。import anyio app.get(/users/{user_id}) async def get_user(user_id: int): result await anyio.to_thread.run_sync(sync_query, user_id) return resultanyio.to_thread.run_sync有一个额外参数abandon_on_cancel默认值是True意思是当外部的协程被取消时线程池里的同步任务不会中断但调用方会直接收到取消异常如果你希望任务被取消后继续在线程池里跑可以设置abandon_on_cancelFalse。这在设计超时控制或后台任务时有帮助。不过我的个人建议是在FastAPI项目里优先用starlette的run_in_threadpool因为它的命名更直白团队协作时其他人一看就懂。anyio.to_thread.run_sync适合在封装底层基础设施时使用比如写一个通用装饰器或工具类。3.3 方案三CPU密集型任务线程池解决不了得上进程池线程池能把同步IO任务从事件循环里挪走但它对CPU密集型任务毫无帮助甚至会拖慢性能。原因是Python的GIL全局解释器锁导致同一进程内同一时刻只有一个线程能执行Python字节码。一个CPU密集型的同步函数丢进线程池虽然不再阻塞事件循环但它占用的还是那块CPU时间片高并发时线程之间互相争夺GIL整体效率反而更差。CPU密集型的典型场景包括图像缩放、OCR识别、机器学习推理、大数据量排序、复杂加密解密。这类任务的正解是进程池from concurrent.futures import ProcessPoolExecutor from fastapi import FastAPI app FastAPI() process_pool ProcessPoolExecutor(max_workers4) def cpu_intensive_task(data: str): # 模拟CPU密集计算 result 0 for i in range(10_000_000): result i return result app.get(/compute) async def compute(): import asyncio loop asyncio.get_running_loop() result await loop.run_in_executor(process_pool, cpu_intensive_task, input) return {result: result}使用进程池时要注意三点一是进程池创建有开销建议在FastAPI lifespan启动时创建、关闭时释放不要每次请求都新建二是传给进程池的任务函数和参数必须能被pickle序列化三是Windows平台下进程池的创建需要放在if __name__ __main__保护块内但FastAPI部署时通常使用uvicorn这个问题影响不大。3.4 更彻底的解法把同步库替换成异步库如果同步方法本身来自一个第三方库而且你想彻底解决阻塞问题最好的办法就是换一个异步实现。最常见的例子是HTTP请求requests是同步库在async def里调用它就必须用线程池兜底但如果你换成httpx它自带AsyncClient可以直接用await收发HTTP请求完全不占线程池。import httpx app.get(/fetch) async def fetch_data(): async with httpx.AsyncClient(timeout10) as client: resp await client.get(https://api.example.com/data) return resp.json()类似的对应关系还有同步文件操作对应aiofiles同步Redis库redis-py对应redis.asyncio同步MySQL驱动PyMySQL对应asyncmy或aiomysql。但换库不是零成本的异步库往往对底层依赖有要求比如asyncmy需要编译、asyncpg和SQLAlchemy的异步驱动搭配要多一些配置。所以我的经验是能用异步库就直接用异步库不能换库或者更换成本太高就老实使用线程池。4. 实战三个典型场景的完整改造过程4.1 场景一异步接口中调用同步HTTP SDK现实项目里很多第三方服务只提供同步SDK比如某些企业内部RPC客户端、云厂商对象存储SDK、旧版消息推送SDK。你无法修改这些SDK的内部实现只能想办法在异步方法里安全调用它们。改造前from fastapi import FastAPI import requests app FastAPI() app.post(/notify) async def notify(user_id: int, message: str): resp requests.post( https://push.internal.example.com/send, json{user_id: user_id, message: message}, timeout5, ) return {status: resp.status_code}这段代码最大的问题requests.post的5秒超时是真实会发生的事件循环阻塞。如果上游推送服务变慢你的FastAPI服务所有接口都会跟着延迟。改造后from fastapi import FastAPI from starlette.concurrency import run_in_threadpool import requests app FastAPI() def _send_push(user_id: int, message: str): resp requests.post( https://push.internal.example.com/send, json{user_id: user_id, message: message}, timeout5, ) return {status: resp.status_code} app.post(/notify) async def notify(user_id: int, message: str): return await run_in_threadpool(_send_push, user_id, message)改造后事件循环不再被requests阻塞。不过要注意这种方式依然受限于线程池大小。如果推送接口QPS特别高、每次响应都慢线程池会被长任务占满这已经属于架构层面问题需要走消息队列异步削峰而不是在线程池里硬扛。4.2 场景二FastAPI SQLAlchemy同步Session在异步路由中使用FastAPI配SQLAlchemy是非常经典的组合但很多教程里的SQLAlchemy用的是同步Session。如果在async def路由里直接操作同步Session就算是简单的session.query(User).filter_by(id1).first()查询期间也会阻塞事件循环。数据库慢查询的时候整个服务的吞吐量会肉眼可见地下降。改造方案有两种。第一种保留同步Session用线程池包装from fastapi import FastAPI from sqlalchemy.orm import Session from starlette.concurrency import run_in_threadpool app FastAPI() def get_user(db: Session, user_id: int): return db.query(User).filter(User.id user_id).first() app.get(/users/{user_id}) async def read_user(user_id: int, db: Session Depends(get_db)): user await run_in_threadpool(get_user, db, user_id) return user这里有个隐含问题db这个同步Session被传进了线程池它不能在多个线程间共享使用。因此依赖项get_db需要保证每个请求拿到的是独立的Session。FastAPI官方文档里常见的get_db写法正好满足这一点它用yield在每个请求生命周期内创建一个新Session用完即关。第二种换成SQLAlchemy 2.0的AsyncSessionfrom fastapi import FastAPI from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine from sqlalchemy import select engine create_async_engine(mysqlasyncmy://user:passwordlocalhost/db) app FastAPI() async def get_db_session(): async with AsyncSession(engine) as session: yield session app.get(/users/{user_id}) async def read_user(user_id: int, db: AsyncSession Depends(get_db_session)): result await db.execute(select(User).where(User.id user_id)) return result.scalar_one_or_none()从长期看AsyncSession是高并发场景下的正确方向因为它把数据库等待也变成了异步IO不占线程池。但从同步Session迁移到AsyncSession所有查询语法都要调整涉及分页、关联查询、事务边界时改动量不小。我个人建议老项目先用run_in_threadpool包装同步查询快速止血新项目直接用异步SQLAlchemy避免二次重构。4.3 场景三后台长时间CPU计算任务有个接口要做PDF解析和文本抽取单次解析耗时十几秒。如果直接把解析函数丢进线程池线程池会被这种长任务占满其他轻量同步任务也跟着排队。正确思路是把这类任务做成后台任务或独立进程接口立即返回任务ID前端轮询进度。用FastAPI自带的BackgroundTasks可以快速实现简单场景但BackgroundTasks实际上还是在线程池里执行CPU密集任务依然受GIL限制。更稳妥的方案是把任务提交给Celery或RQ这样的独立任务队列。如果不想引入消息队列也可以直接用进程池 任务表from fastapi import BackgroundTasks from concurrent.futures import ProcessPoolExecutor process_pool ProcessPoolExecutor(max_workers2) def heavy_parse_pdf(file_path: str) - str: # 真正的CPU密集解析 time.sleep(15) return parsed_content app.post(/parse) async def parse_pdf(file_path: str, background_tasks: BackgroundTasks): loop asyncio.get_running_loop() background_tasks.add_task(loop.run_in_executor, process_pool, heavy_parse_pdf, file_path) return {task_id: submitted}这里要注意BackgroundTasks.add_task的第一个参数是函数后面的参数是传给函数的参数。把loop.run_in_executor作为任务函数有点绕但确实可行因为run_in_executor本身是协程BackgroundTasks会等待它完成。核心思想是CPU密集型任务不占事件循环、不占IO线程池而是占独立进程池。4.4 设计建议把异步边界画在项目架构里实战用过一段时间后我的体会是异步调用同步不再是一个“代码技巧”问题而是项目架构问题。FastAPI项目目录结构如果设计得好这种混用会非常自然。我的习惯是app/ main.py # FastAPI实例、CORS配置、路由注册、lifespan api/ # 路由层全部是async def只做参数校验和响应组装 services/ # 业务逻辑层可以同时有同步函数和异步函数 clients/ # 外部服务封装同步SDK统一封装为线程池版接口 db/ # 数据库Session管理、模型定义、CRUD路由层保持async def业务层的同步函数由路由层通过run_in_threadpool调用外部同步SDK封装到clients目录里全部对外暴露成异步接口这样路由层永远不用关心底层是同步还是异步。CORS配置放在main.py的FastAPI实例上跨域问题和并发问题分开处理不要让它们纠缠在一起。这样做还有一个好处当某个同步SDK升级到异步版本时只需要改clients目录里的封装路由层和业务层完全不用动影响范围控制得非常小。5. 常见问题与排查技巧实录5.1 用了run_in_threadpool还是慢并发上限卡在哪如果你已经把所有同步调用都包进了run_in_threadpool但压测时发现并发到了某个值后延迟陡增最可能的问题是线程池打满了。asyncio默认线程池的max_workers是min(32, os.cpu_count() 4)对于大量IO密集任务来说这个线程数很容易被占满。可以自定义线程池容量根据业务并发量估算。比如100个线程import asyncio from concurrent.futures import ThreadPoolExecutor # 给事件循环设置自定义线程池 loop asyncio.get_event_loop() loop.set_default_executor(ThreadPoolExecutor(max_workers100))但设置线程数不是越大越好。线程太多会带来上下文切换开销和内存压力通常100到200是常见区间超过这个量就应该考虑异步化替换或消息队列。还有一个更隐蔽的问题线程池里的同步任务如果在等待某个共享资源比如同一个数据库连接池耗尽、同一个文件锁被占用线程池满不一定是线程不够而是资源被卡住了排查时要结合连接的等待时长一起看。5.2 Event loop is closed看到这个错误别慌RuntimeError: Event loop is closed在FastAPI异步场景里出现频率不低。常见原因有几个在测试环境中用asyncio.run启动了一个协程但这个协程内部用asyncio.get_event_loop拿到了另一个loop测试结束后旧loop被关闭。使用uvicorn--reload时旧代码里的后台任务在新loop中访问了旧loop的资源。某个对象比如数据库连接绑定了创建时的loop后续被另一个loop使用。排查思路先定位报错时的调用栈看看是不是跨loop持有了某些连接对象。通常解决办法是在FastAPI的lifespan里统一管理资源的创建和销毁不要在每个函数里手动创建loop。from contextlib import asynccontextmanager import httpx asynccontextmanager async def lifespan(app: FastAPI): # 在lifespan里创建全局异步客户端 app.state.client httpx.AsyncClient() yield # 关闭 await app.state.client.aclose() app FastAPI(lifespanlifespan)5.3 函数明明写了async为什么仍然卡一种情况是async函数内部没有真正的await所有同步操作都在事件循环里裸跑这在前面已经说过。另一种情况是函数内部虽然有await但await的对象是一个同步IO封装。比如# 下面这段其实是同步阻塞 await一个普通对象性能很差 async def get_data(): data some_sync_call() # 阻塞 await asyncio.sleep(0.01) # 只是象征性挂起 return data判断一个async函数有没有真正异步化可以在压测时观察事件循环延迟或者看CPU单核使用率是否接近100%。如果单核拉满但并发上不去说明同步阻塞代码还在事件循环里。这块没有快速捷径只能逐个检查耗时操作是否被正确封装。5.4 如何快速定位到底哪个同步调用阻塞了事件循环项目大的时候从几十个async函数里找出偷偷阻塞的那个很费劲。分享两个我常用的办法。第一个是py-spy通过py-spy dump --pid uvicorn进程号可以打印Python进程里所有线程的当前调用栈。事件循环线程的栈上如果停留在一个明显的同步调用上比如socket连接、文件读取那就是罪魁祸首。第二个是观察法如果某个接口或定时任务运行期间其他异步请求明显卡顿可以在可疑函数前后打印时间戳逐步缩小范围。这类问题最好在开发阶段就建立拦截机制比如封装一个safe_call函数强制所有同步调用都走run_in_threadpool从源头杜绝裸调。现象可能原因优先级单个接口响应慢其他接口正常该接口内部有阻塞调用高所有async接口都变慢def接口正常事件循环被阻塞高所有接口都变慢包括def接口线程池或数据库连接池耗尽中压测时CPU单核100%CPU密集任务跑在事件循环或线程池中报错Event loop is closed跨loop持有资源低做了一段时间FastAPI开发后最深的感受是异步方法里调用同步方法这个事越小越容易被忽视但危害往往在流量上来之后才暴露。我现在写代码的基本纪律是async def函数里出现的每一个可能阻塞的调用都要过一遍脑子能用异步库就用异步库不能就丢线程池绝不裸调。这个习惯一开始会觉得繁琐时间久了反而会让代码的可维护性提高不少。最后再分享一个小技巧给写接口的新人做Code Review时直接搜async def函数体内是否还有未经过await run_in_threadpool包裹的同步IO调用这个检查比任何规范文档都管用。