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

前言

在异步方法中调用同步方法，会直接壅闭整个事故循环，导致应用在实行同步方法期间无法处理处罚其他任何并发哀求，从而拖垮整个服务的性能。
为了办理这个标题，核心思绪是将同步方法交给外部线程池去实行。
方法1, 使用 to_thread

Python 3.9 后可以使用 asyncio.to_thread 方法，将同步函数跑在独立的线程中，并返回一个协程供 await

1. import asyncio
2. import time
3. from fastapi import FastAPI
5. app = FastAPI()
7. def sync_task(name: str):
8.     time.sleep(2)
9.     return f"Hello {name}, sync task done!"
11. @app.get("/async-call")
12. async def async_endpoint():
13.     result = await asyncio.to_thread(sync_task, "World")
14.    
15.     return {"message": result}

复制代码

方法2, 直接界说同步路由

FastAPI支持界说同步路由，FastAPI会自动在一个外部线程池中运行该函数。不外出于代码团体筹划的思量，个人不发起这么做。
方法3, 使用 run_in_threadpool

FastAPI 基于 Starlette, 而 Starlette 提供一个工具函数 run_in_threadpool，这种方式类似于 asyncio.to_thread，在某些老版本的 FastAPI 或特定的 contextvars 通报场景下更常用。

1. from fastapi.concurrency import run_in_threadpool
3. @app.get("/method3")
4. async def starlette_endpoint():
5.     result = await run_in_threadpool(sync_task, "Starlette")
6.     return {"message": result}

复制代码

方法4, 使用历程池

对于CPU麋集型任务，应该使用多历程ProcessPoolExecutor来操纵

1. import concurrent.futures
2. import math
3. from fastapi import FastAPI
5. app = FastAPI()
6. # 创建一个全局进程池
7. executor = concurrent.futures.ProcessPoolExecutor()
9. def cpu_intensive_calculation(n: int):
10.     # 模拟重度 CPU 计算
11.     return sum(math.isqrt(i) for i in range(n))
13. @app.get("/cpu-bound-task")
14. async def cpu_task():
15.     loop = asyncio.get_running_loop()
16.     result = await loop.run_in_executor(executor, cpu_intensive_calculation, 10**7)
17.     return {"result": result}

复制代码