如何限制FastAPI一次仅处理一个请求，且处理期间对其他请求返回“服务器繁忙”响应

刚好之前做过类似的需求，FastAPI里要实现「同一时间仅处理一个请求，其余请求返回‘服务器繁忙’」其实很简单，核心就是用全局锁+自定义依赖来做请求拦截。下面分不同场景给你具体实现：

一、异步路由场景 #

如果你的接口都是异步实现的，用asyncio.Lock来控制并发最适合：

from fastapi import FastAPI, Depends, HTTPException
import asyncio

app = FastAPI()

# 全局异步锁，确保同一时间只有一个异步请求被处理
_global_async_lock = asyncio.Lock()

async def acquire_async_lock():
    # 尝试立即获取锁，不等待
    if not _global_async_lock.locked():
        await _global_async_lock.acquire()
        try:
            yield  # 执行路由逻辑
        finally:
            _global_async_lock.release()  # 无论成功失败都释放锁
    else:
        # 锁已被占用，直接返回503繁忙提示
        raise HTTPException(status_code=503, detail="服务器繁忙")

@app.get("/async-heavy-task")
async def run_async_heavy_task(_=Depends(acquire_async_lock)):
    # 模拟耗时异步任务（比如调用第三方API、IO操作）
    await asyncio.sleep(5)
    return {"status": "success", "message": "异步耗时任务执行完成"}

二、同步路由场景 #

如果你的接口是同步实现的（比如用CPU密集型任务），需要用threading.Lock来控制线程池里的并发：

from fastapi import FastAPI, Depends, HTTPException
import threading
import time

app = FastAPI()

# 全局线程锁，控制同步请求的并发
_global_sync_lock = threading.Lock()

def acquire_sync_lock():
    # 非阻塞式获取锁
    if _global_sync_lock.acquire(blocking=False):
        try:
            yield
        finally:
            _global_sync_lock.release()
    else:
        raise HTTPException(status_code=503, detail="服务器繁忙")

@app.get("/sync-heavy-task")
def run_sync_heavy_task(_=Depends(acquire_sync_lock)):
    # 模拟耗时同步任务（比如数据计算、文件处理）
    time.sleep(5)
    return {"status": "success", "message": "同步耗时任务执行完成"}

三、混合场景（异步+同步路由） #

如果你的项目同时有异步和同步接口，推荐把同步任务包装成异步形式，统一用asyncio.Lock控制，避免维护两种锁：

from fastapi import FastAPI, Depends, HTTPException
import asyncio
import time

app = FastAPI()

_global_lock = asyncio.Lock()

async def acquire_unified_lock():
    if not _global_lock.locked():
        await _global_lock.acquire()
        try:
            yield
        finally:
            _global_lock.release()
    else:
        raise HTTPException(status_code=503, detail="服务器繁忙")

# 异步接口
@app.get("/async-task")
async def async_task(_=Depends(acquire_unified_lock)):
    await asyncio.sleep(5)
    return {"message": "异步任务完成"}

# 同步接口包装成异步
@app.get("/sync-task")
async def sync_task(_=Depends(acquire_unified_lock)):
    # 用asyncio.to_thread把同步任务放到线程池执行
    await asyncio.to_thread(lambda: time.sleep(5))
    return {"message": "同步任务完成"}

关键注意事项 #

• 单进程限制：上面的代码都是单进程内的控制，如果你的FastAPI用了多进程部署（比如uvicorn --workers 2），每个进程会有独立的锁，此时需要用分布式锁（比如Redis锁）来跨进程控制并发。
• 非阻塞获取：代码里用blocking=False是核心，这样请求不会排队等待，而是直接返回繁忙提示，完全符合需求。如果设为blocking=True，请求会排队等待锁释放，就达不到“返回繁忙”的效果了。
• 锁的释放：用try...finally确保锁一定会被释放，避免因为路由抛出异常导致死锁。

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