如何在无重型任务时定期重启带上下文管理器的FastAPI

数据老陈

2026-6-30

解决方案

核心思路是将FastAPI应用与外部库的上下文管理器解耦，不再让uvicorn直接运行包裹在上下文里的应用，而是动态管理上下文实例，同时跟踪任务运行状态，确保仅在空闲时重启上下文。

具体实现步骤

1. 封装上下文管理逻辑

把外部库的上下文初始化、销毁逻辑封装成可重复调用的函数，方便重启时重新创建实例：

from contextlib import contextmanager
import uvicorn
from fastapi import FastAPI, Depends
import time
from threading import Lock

# 模拟外部库的上下文管理器（替换成你的实际库逻辑）
@contextmanager
def required_context_manager():
    print("初始化外部库上下文")
    # 这里写外部库的初始化代码，比如连接资源、加载配置等
    yield
    print("销毁外部库上下文")
    # 这里写外部库的清理代码，比如释放资源、关闭连接等

# 全局变量存储当前上下文实例、任务计数，加锁避免多线程竞争
current_context = None
active_heavy_tasks = 0
task_lock = Lock()

2. 实现上下文的动态获取与初始化

通过依赖注入让路由能访问当前有效的上下文，同时提供初始化/重启上下文的函数：

# 路由依赖：获取当前生效的上下文
def get_current_context():
    global current_context
    if not current_context:
        raise RuntimeError("外部库上下文未初始化")
    return current_context

# 初始化/重启上下文的函数
def refresh_context():
    global current_context
    # 先销毁旧上下文（如果存在）
    if current_context:
        current_context.__exit__(None, None, None)
    # 创建并激活新上下文
    ctx = required_context_manager()
    current_context = ctx.__enter__()
    return current_context

3. 跟踪重型任务状态

给所有重型任务路由加装饰器，统计当前运行的重型任务数量，确保重启时系统处于空闲：

# 装饰器：跟踪重型任务的启动与结束
def track_heavy_task(func):
    def wrapper(*args, **kwargs):
        global active_heavy_tasks
        with task_lock:
            active_heavy_tasks += 1
        try:
            return func(*args, **kwargs)
        finally:
            with task_lock:
                active_heavy_tasks -= 1
    return wrapper

4. 实现自动/手动重启机制

后台线程定时检查，空闲时自动重启上下文
提供内部接口，允许手动触发重启（用于调试或紧急场景）

app = FastAPI()

# 示例重型任务路由（所有重型任务都要加@track_heavy_task装饰器）
@app.get("/process-large-data")
@track_heavy_task
async def process_large_data(context=Depends(get_current_context)):
    # 模拟重型任务执行
    time.sleep(10)
    return {"status": "completed", "context_instance_id": id(context)}

# 手动触发上下文重启的接口（建议加权限控制，仅内部调用）
@app.get("/admin/restart-context")
async def manual_restart_context():
    with task_lock:
        if active_heavy_tasks > 0:
            return {"status": "failed", "reason": "当前有重型任务运行，无法重启"}
    refresh_context()
    return {"status": "success", "new_context_id": id(current_context)}

# 定时重启线程：每隔1小时检查一次，空闲则重启
def scheduled_context_refresh(interval=3600):
    while True:
        time.sleep(interval)
        with task_lock:
            if active_heavy_tasks == 0:
                print("检测到系统空闲，执行上下文重启")
                refresh_context()
            else:
                print("当前有任务运行，跳过本次重启")

5. 启动服务

初始化第一个上下文，启动定时线程，再启动uvicorn：

if __name__ == "__main__":
    # 初始化第一个上下文
    refresh_context()
    # 启动定时重启后台线程（设为守护线程，随主进程退出）
    import threading
    restart_thread = threading.Thread(target=scheduled_context_refresh, args=(3600,), daemon=True)
    restart_thread.start()
    # 启动FastAPI服务（不需要包裹在上下文管理器里）
    uvicorn.run(app, host="0.0.0.0", port=8000)

关键注意事项

禁止使用uvicorn的--reload选项，避免多线程导致上下文管理失效
如果外部库是异步上下文管理器，需将refresh_context改成异步函数，定时线程换成异步任务（比如用asyncio.create_task）
全局变量的操作必须加锁，防止多线程下的竞争问题
手动重启接口建议添加身份验证，避免未授权访问

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