如何在FastAPI中确保异步函数同一时间仅运行一次？

1. 修复异步任务的阻塞问题 #

你的asyncio_test_prediction里用了time.sleep(30)，这是同步阻塞操作，会卡住整个asyncio事件循环，导致其他请求也被阻塞。必须换成异步的asyncio.sleep：

import asyncio
from fastapi import FastAPI

app = FastAPI()

async def asyncio_test_prediction():
    print("Starting asyncio func")
    await asyncio.sleep(30)  # 替换time.sleep为asyncio.sleep
    print("Stopping asyncio func")

2. 存储任务实例并检查运行状态 #

我们可以用FastAPI的app.state来存储任务实例，这样能在整个应用生命周期中跟踪任务状态。每次调用接口时，先检查任务是否存在且未完成：

异步接口实现（推荐） #

# 在app初始化后添加任务存储变量
app.state.prediction_task = None

@app.get('/sempos/start')
async def start_prediction():
    # 检查任务是否存在且未完成
    if app.state.prediction_task is not None and not app.state.prediction_task.done():
        return {"state": "Already running"}
    
    # 创建新任务并存储
    app.state.prediction_task = asyncio.create_task(asyncio_test_prediction())
    return {"state": "Started"}

同步接口实现 #

如果你的接口不需要异步逻辑，也可以写成同步函数，此时需要手动获取当前事件循环来创建任务：

@app.get('/sempos/start')
def start_prediction():
    loop = asyncio.get_running_loop()
    if app.state.prediction_task is not None and not app.state.prediction_task.done():
        return {"state": "Already running"}
    
    app.state.prediction_task = loop.create_task(asyncio_test_prediction())
    return {"state": "Started"}

3. 额外说明 #

• asyncio.create_task是Python 3.7+推荐的创建异步任务的方式，比asyncio.ensure_future更直观。
• 任务完成后，task.done()会返回True，下次调用接口时会自动创建新任务。
• 如果需要处理任务的返回值或者异常，可以给任务添加回调函数，比如：

  def task_callback(task):
      try:
          result = task.result()
          print(f"Task result: {result}")
      except Exception as e:
          print(f"Task failed: {e}")
  
  # 创建任务时绑定回调
  task = asyncio.create_task(asyncio_test_prediction())
  task.add_done_callback(task_callback)
  

内容的提问来源于stack exchange，提问作者Tropaion