在 FastAPI 异步函数中调用同步阻塞代码会导致事件循环阻塞，引发服务超时。优化方案核心是将同步操作移至线程池或进程池执行。具体可使用 FastAPI 提供的 run_in_threadpool 工具，或 Python 3.9+ 的 asyncio.to_thread 方法。对于 IO 密集型任务，建议替换为异步库（如 httpx 替代 requests）；对于 CPU 密集型任务，应使用进程池。避免在 async def 中直接使用 time.sleep 或同步数据库连接，确保事件循环不被占用，从而提升并发处理能力。

Fastapi 服务在高并发情况下大量超时问题排查

通过排查，找到了代码中的问题：在一个异步方法中调用了一个同步方法，这就导致一个容器服务一次只能支持一个并发。我们直接用代码进行演示。代码演示 1. 服务代码 importtimeimportasynciofromfastapiimportFastAPIfromfastapi.concurrencyimportrun_in_threadpool app=FastAPI()# 模拟一个耗时的同步函数 (比如：处理大文件、调用老系统 API、复杂计算)defslow_sync_task(task_id:int):print(f"Task{task_id}started (blocking for 3 seconds)")time.sleep(3)# 这是阻塞操作！会卡住整个事件循环 print(f"Task{task_id}finished.")returnf"Result from task{task_id}"# ❌ 错误方式：在 async 接口中直接调用同步方法@app.get("/bad")asyncdefbad_endpoint(task_id:int=1):# 直接调用同步阻塞函数 result=slow_sync_task(task_id)return{"method":"bad","result":result}# ✅ 正确方式 1:使用 FastAPI 的 run_in_threadpool@app.get("/good1")asyncdefgood_endpoint_v1(task_id:int=1):# 将同步函数放到线程池中执行，不阻塞事件循环 result=awaitrun_in_threadpool(slow_sync_task,task_id)return{"method":"good1","result":result}# ✅ 正确方式 2:使用 asyncio.to_thread (Python 3.9+ 推荐)@app.get("/good2")asyncdefgood_endpoint_v2(task_id:int=1):# 同样是非阻塞的，更现代的写法 result=awaitasyncio.to_thread(slow_sync_task,task_id)return{"method":"good2","result":result}# ✅ 额外对比：真正的异步非阻塞 (比如模拟网络请求)asyncdeffake_async_io_task(task_id:int):print(f"Async Task{task_id}started (waiting 3 seconds asynchronously)")awaitasyncio.sleep(3)# 这是真正的异步等待，不会阻塞 print(f"Async Task{task_id}finished.")returnf"Async result from task{task_id}"@app.get("/perfect")asyncdefperfect_endpoint(task_id:int=1):result=awaitfake_async_io_task(task_id)return{"method":"perfect","result":result}if__name__=="__main__":importuvicorn uvicorn.run(app,host="0.0.0.0",port=8000)

FastAPI 异步方法中调用同步方法

FastAPI 异步方法中调用同步方法 前言 在异步方法中调用同步方法，会直接阻塞整个事件循环，导致应用在执行同步方法期间无法处理其他任何并发请求，从而拖垮整个服务的性能。为了解决这个问题，核心思路是将同步方法交给外部线程池去执行。方法 1, 使用 to_thread Python 3.9 后可以使用 asyncio.to_thread 方法，将同步函数跑在独立的线程中，并返回一个协程供 await importasyncio importtime fromfastapiimportFastAPI app = FastAPI() defsync_task(name:str): time.sleep(2) returnf"Hello{name}, sync task done!" @app.get("/async-call") asyncdefasync_endpoint(): result =awaitasyncio.to_thread(sync_task,"World") return{"message": result} 一键获取完整项目代码 方法 2, 直接定义同步路由 FastAPI 支持定义同步路由，FastAPI 会自动在一个外部线程池中运行该函数。不过出于代码整体设计的考虑，个人不建议这么做。方法 3, 使用 run_in_threadpool FastAPI 基于 Starlette, 而 Starlette 提供一个工具函数 run_in_threadpool，这种方式类似于 asyncio.to_thread，在某些老版本的 FastAPI 或特定的 contextvars 传递场景下更常用。fromfastapi.concurrencyimportrun_in_threadpool @app.get("/method3") asyncdefstarlette_endpoint(): result =awaitrun_in_threadpool(sync_task,"Starlette") return{"message": result} 一键获取完整项目代码 方法 4, 使用进程池 对于 CPU 密集型任务，应该使用多进程 ProcessPoolExecutor 来操作 importconcurrent.futures importmath fromfastapiimportFastAPI app = FastAPI() # 创建一个全局进程池 executor = concurrent.futures.ProcessPoolExecutor() defcpu_intensive_calculation(n:int): # 模拟重度 CPU 计算 returnsum(math.isqrt(i)foriinrange(n)) @app.get("/cpu-bound-task") asyncdefcpu_task(): loop = asyncio.get_running_loop() result =awaitloop.run_in_executor(executor, cpu_intensive_calculation,10**7) return{"result": result} 一键获取完整项目代码

FastApi 框架异步调用同步问题 - 輪滑少年 - 博客园

Fastapi 项目，在接口中调用同步方法，如果该同步方法，耗时较长 (比如连接 redis 超时),会造成整个项目接口的阻塞，这是任何接口的访问都会被阻塞超时 一、为什么会阻塞 FastAPI 是基于异步框架 (如 asyncio 或 anyio) 构建的，它的核心是一个事件循环 (Event Loop)。事件循环负责调度和执行所有的异步任务。当你在异步函数中直接调用同步阻塞代码时，事件循环会被阻塞，无法继续处理其他任务，直到同步代码执行完毕。fromfastapiimportFastAPIimporttime app=FastAPI()defsync_function(): time.sleep(5)#模拟一个耗时的同步操作 return"Done"@app.get("/") asyncdefroot(): result= sync_function()#直接调用同步函数 return{"result": result} 在这个例子中 time.sleep(5) 是一个同步阻塞操作，它会阻塞事件循环 5 秒钟。在这期间，FastAPI 无法处理其他请求，整个应用的并发性能会大幅下降。二、如何避免阻塞 将同步代码放到线程池或进程池中执行，将同步代码改为异步实现。有以下几种常用解决方案 1. 使用 run_in_threadpool 或 asyncio.to_thread 将同步代码放到线程池中执行，避免阻塞事件循环。fromfastapiimportFastAPIfromfastapi.concurrencyimportrun_in_threadpoolimporttime app=FastAPI()defsync_function(): time.sleep(5)#模拟一个耗时的同步操作 return"Done"@app.get("/") asyncdefroot(): result= await run_in_threadpool(sync_function)#在线程池中运行 return{"result": result} 2 使用 run_in_processpool 对于 CPU 密集型的同步代码，可以使用进程池来避免阻塞事件循环。fromfastapiimportFastAPIfromfastapi.concurrencyimportrun_in_processpoolimporttime app=FastAPI()defcpu_intensive_function():#模拟一个 CPU 密集型的操作 result = sum(i * iforiinrange(10**6))returnresult @app.get("/") asyncdefroot(): result= await run_in_processpool(cpu_intensive_function)#在进程池中运行 return{"result": result} 3. 将同步代码改为异步实现 如果可能，尽量将同步代码改为异步实现。例如，使用 asyncio.sleep 代替 time.sleep，或者使用异步库代替同步库。fromfastapiimportFastAPIimportasyncio app=FastAPI() asyncdefasync_function(): await asyncio.sleep(5)#异步等待 return"Done"@app.get("/") asyncdefroot(): result= await async_function()#直接调用异步函数 return{"result": result} 4. 使用 anyio.to_thread.run_sync

Python FastAPI 异步怎么阻塞了_由于在 async def 中运行了计算密集或同步网络请求导致卡顿

FastAPI 的 async def 路由中禁用 requests.get、time.sleep()、sqlite3.connect() 等同步阻塞操作;应改用 httpx.AsyncClient、await asyncio.sleep()、aiosqlite 或 asyncpg;CPU 密集任务需用 asyncio.to_thread 或 run_in_executor;并发需用 asyncio.gather 或 create_task，避免串行 await。async def 里调用了 requests.get 就卡住 FastAPI 的 async def 路由函数只在事件循环中“不阻塞 I/O",但 requests 是同步阻塞式 HTTP 客户端，一调用就停住整个协程线程，其他请求全得排队等它返回。常见错误现象：接口响应时间忽高忽低，压测时并发上不去，uvicorn 日志里看不到并发请求日志，像单线程在跑。换成异步 HTTP 客户端:httpx.AsyncClient(推荐) 或 aiohttp 如果必须用 requests，得用 loop.run_in_executor 包一层，丢给线程池执行 httpx 默认支持 HTTP/2 和连接复用，比 requests 在高并发下更稳 复制 AI 写代码 import httpx @app.get("/fetch") async def fetch_data(): async with httpx.AsyncClient() as client: resp = await client.get("https://api.example.com/data") return resp.json() CPU 密集任务写在 async def 里没用 比如做图像缩放、JSON 大文件解析、加密解密、正则全文匹配——这些根本不是 I/O，而是纯 CPU 运算，async 不会加速，反而因协程调度增加开销。使用场景：你有个 process_large_csv() 函数，输入是文件路径，输出是处理后的字典列表，但它内部用 pandas.read_csv+.apply() 做大量计算。别在 async def 里直接调用，哪怕加了 await 也没意义 用 asyncio.to_thread()(Python 3.9+) 或 loop.run_in_executor(None, ) 拆出去 注意：频繁小任务用线程池反而有上下文切换成本，大任务才值得拆 复制 AI 写代码 import asyncio @app.post("/analyze") async def analyze_text(data: dict): result = await asyncio.to_thread(process_heavy_nlp, data["text"]) return {"result": result} async def 路由里混用 time.sleep() 或 sqlite3.connect() 这两个是典型“看着像 I/O，实则完全阻塞”的坑。time.sleep() 是纯同步休眠;sqlite3 默认也是同步磁盘操作，即使开 isolation_level=None 也挡不住锁和写入阻塞。错误现象：一个请求 sleep 2 秒，其他所有请求都得等;多个请求同时写 sqlite，出现 database is locked 错误且响应超时。time.sleep() 必须换 await asyncio.sleep() SQLite 不适合高并发写，要么加 WAL 模式 + 连接池 (aiosqlite)，要么切到异步数据库如 asyncpg(PostgreS

FAQ

为什么在 FastAPI 异步函数中直接调用同步方法会导致超时？

因为 FastAPI 基于事件循环，同步方法会阻塞事件循环，导致其他请求无法处理。

对于 CPU 密集型任务应该使用线程池还是进程池？

对于 CPU 密集型任务，应该使用多进程 ProcessPoolExecutor 来操作，避免阻塞事件循环。

在 FastAPI 中如何进行 HTTP 请求以避免阻塞？

应改用 httpx.AsyncClient 或 aiohttp 等异步 HTTP 客户端，避免使用同步的 requests 库。