FastAPI 异步后台任务阻塞其他请求？

Question

我想在 FastAPI 中运行一个简单的后台任务，这涉及在将其转储到数据库之前进行一些计算。但是，计算会阻止它接收更多请求。

 from fastapi import BackgroundTasks, FastAPI

app = FastAPI()
db = Database()

async def task(data):
    otherdata = await db.fetch("some sql")
    newdata = somelongcomputation(data,otherdata) # this blocks other requests
    await db.execute("some sql",newdata)


@app.post("/profile")
async def profile(data: Data, background_tasks: BackgroundTasks):
    background_tasks.add_task(task, data)
    return {}

解决此问题的最佳方法是什么？

原文由 Gary Ong 发布，翻译遵循 CC BY-SA 4.0 许可协议

Answer 1

您的 task 定义为 async ，这意味着 fastapi（或者更确切地说是 starlette）将在 asyncio 事件循环中运行它。并且因为 somelongcomputation 是同步的（即不等待某些 IO，而是进行计算）只要它正在运行，它就会阻塞事件循环。

我看到了解决这个问题的几种方法：

• 使用更多的工人（例如 uvicorn main:app --workers 4 ）。这将允许最多 4 个 somelongcomputation 并行。

• 将您的任务重写为不是 async （即将其定义为 def task(data): ... 等）。然后 starlette 将在单独的线程中运行它。

• 使用 fastapi.concurrency.run_in_threadpool ，这也将在单独的线程中运行它。像这样：

   from fastapi.concurrency import run_in_threadpool
  async def task(data):
      otherdata = await db.fetch("some sql")
      newdata = await run_in_threadpool(lambda: somelongcomputation(data, otherdata))
      await db.execute("some sql", newdata)

• 或者直接使用 asyncios 的 run_in_executor （其中 run_in_threadpool 在后台使用）：

   import asyncio
  async def task(data):
      otherdata = await db.fetch("some sql")
      loop = asyncio.get_running_loop()
      newdata = await loop.run_in_executor(None, lambda: somelongcomputation(data, otherdata))
      await db.execute("some sql", newdata)
  
  
  

  您甚至可以传入 concurrent.futures.ProcessPoolExecutor 作为 run_in_executor 的第一个参数，以在单独的进程中运行它。

• 自己生成一个单独的线程/进程。例如使用 concurrent.futures 。

• 使用更重的东西，如芹菜。 （也在 此处 的 fastapi 文档中提到）。

原文由 mihi 发布，翻译遵循 CC BY-SA 4.0 许可协议

Answer 2

如果您的任务受 CPU 限制，您可以使用多处理，可以使用 FastAPI 中的后台任务执行此操作： https ://stackoverflow.com/a/63171013

虽然如果有很多 cpu 繁重的任务，你应该考虑使用像 Celery 这样的东西。

原文由 Adam 发布，翻译遵循 CC BY-SA 4.0 许可协议