FastAPI路由单次调用实现：执行中阻止重复调用并提示

你用BackgroundTasks搞不定状态跟踪很正常，因为它本身没提供这个功能，得自己手动维护任务的运行状态。下面给你两个实用方案：

单进程部署方案（简单直接） #

如果你的服务是单进程跑的，用个全局变量就能搞定状态跟踪：

from fastapi import FastAPI, BackgroundTasks
import time

app = FastAPI()

# 全局变量标记任务是否在运行
task_running = False

def long_running_task():
    global task_running
    try:
        # 这里替换成你的实际耗时操作，比如数据处理、文件生成
        time.sleep(10)
    finally:
        # 不管任务成功失败，最后都要把状态改回去
        task_running = False

@app.post("/run-task")
async def run_task(background_tasks: BackgroundTasks):
    global task_running
    if task_running:
        return {"message": "task in process"}
    # 标记任务开始运行
    task_running = True
    background_tasks.add_task(long_running_task)
    return {"message": "task started successfully"}

原理很简单：每次调用路由先查task_running，如果是True就返回提示；否则标记为运行中，把耗时任务丢给BackgroundTasks去后台跑，任务结束后自动重置状态。要是有多个不同路由需要保护，把全局变量改成字典就行，比如running_tasks = {"task_a": False, "task_b": False}，对应不同路由的状态。

多进程/分布式部署方案（用Redis共享状态） #

要是你的服务用了多进程或者分布式部署，全局变量就没用了——每个进程有自己的内存空间，状态不共享。这时候得用外部存储，Redis是最常用的：

from fastapi import FastAPI, BackgroundTasks
import redis
import time

app = FastAPI()
# 初始化Redis客户端，根据你的实际配置改参数
redis_client = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)
# 用一个唯一的键来标记任务状态
TASK_STATUS_KEY = "my_service_running_task"

def long_running_task():
    try:
        # 替换成你的实际耗时操作
        time.sleep(10)
    finally:
        # 任务结束后删除Redis里的状态标记
        redis_client.delete(TASK_STATUS_KEY)

@app.post("/run-task")
async def run_task(background_tasks: BackgroundTasks):
    if redis_client.exists(TASK_STATUS_KEY):
        return {"message": "task in process"}
    # 设置状态键，同时加个过期时间（比如30秒），防止任务意外崩溃导致状态一直锁着
    redis_client.setex(TASK_STATUS_KEY, 30, "running")
    background_tasks.add_task(long_running_task)
    return {"message": "task started successfully"}

这里用Redis的键来做分布式锁，所有进程都能读到同一个状态。设置过期时间是关键——万一任务因为异常直接挂了，Redis会自动删掉这个键，不会导致后续请求一直被挡。

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