如何在FastAPI应用空闲时运行额外代码（非线程/多进程）

一、异步启动FastAPI应用（非阻塞方式） #

不用uvicorn.run()的阻塞调用，你可以直接使用Uvicorn的Server和Config类，在异步函数中启动服务器，这样就能在同一个事件循环里同时运行服务器和其他后台任务：

import asyncio
from fastapi import FastAPI
from uvicorn import Server, Config

app = FastAPI()

@app.get("/")
async def read_root():
    return {"Hello": "World"}

async def background_idle_task():
    # 后续实现空闲检测逻辑的地方
    while True:
        await asyncio.sleep(5)  # 定期检查间隔
        print("检查是否空闲...")

async def main():
    # 配置FastAPI应用
    config = Config(app=app, host="0.0.0.0", port=8000)
    server = Server(config)
    
    # 同时运行服务器和后台任务
    await asyncio.gather(server.serve(), background_idle_task())

if __name__ == "__main__":
    asyncio.run(main())

这种方式启动后，服务器和后台任务会在同一个事件循环里异步执行，不会互相阻塞。

二、检测空闲状态并执行代码 #

FastAPI没有内置的空闲检测机制，但可以通过中间件+后台任务的方式实现，核心思路是跟踪当前正在处理的请求数量：

完整实现代码 #

import asyncio
from fastapi import FastAPI, Request
from uvicorn import Server, Config

app = FastAPI()

# 全局变量跟踪活跃请求数，用asyncio.Lock保证异步环境下的线程安全
active_requests = 0
request_lock = asyncio.Lock()

# 中间件：统计活跃请求数
@app.middleware("http")
async def track_active_requests(request: Request, call_next):
    global active_requests
    async with request_lock:
        active_requests += 1
    try:
        response = await call_next(request)
    finally:
        async with request_lock:
            active_requests -= 1
    return response

@app.get("/")
async def read_root():
    await asyncio.sleep(2)  # 模拟请求处理耗时
    return {"Hello": "World"}

async def idle_task():
    while True:
        await asyncio.sleep(3)  # 每3秒检查一次空闲状态
        async with request_lock:
            current_active = active_requests
        if current_active == 0:
            print("应用处于空闲状态，执行空闲任务...")
            # 在这里编写你要在空闲时运行的代码
            # 例如清理缓存、同步离线数据等
            await asyncio.sleep(1)  # 模拟空闲任务耗时

async def main():
    config = Config(app=app, host="0.0.0.0", port=8000)
    server = Server(config)
    await asyncio.gather(server.serve(), idle_task())

if __name__ == "__main__":
    asyncio.run(main())

核心说明 #

• 用asyncio.Lock保护active_requests变量，避免异步环境下的竞态问题
• 中间件在请求进入时递增计数，请求处理完成后递减计数
• 后台任务定期检查计数，当计数为0时触发空闲逻辑
• 所有逻辑都在同一个异步事件循环中执行，无需使用threading或multiprocessing模块

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