FastAPI中如何安全重载所有Worker进程状态？附实现需求

核心需求 #

在FastAPI中实现无需重启服务，定期从磁盘重载字典，让所有Worker进程和线程安全感知到变化，允许Worker短暂不同步。同时明确FastAPI后台任务的运行方式。

现有代码示例 #

启动代码：

some_dict = load_data(path)
server = Server(some_dict)
app = FastAPI()  
app.include_router(server.router)
uvicorn.run(app, ...)

Server类：

class Server:
   def __init__(self, some_dict: Dict[str, ...]):
       self.router = APIRouter()
       self.router.add_api_route("/bar", self.bar, methods=["POST"])
       self.some_dict = some_dict 

   def bar(self, text: str):
       pass

具体疑问 #

• 是否有办法定期从磁盘重载该字典，让所有Worker和线程都能安全感知到变化？可使用或不使用FastAPI内置功能。
• FastAPI的后台任务是运行在每个Worker进程中，还是单独的进程？
• 希望通过构建新字典实例并替换引用的方式实现重载（如下示例），求最简实现：

# this should be the only place to update state of some_dict
class Server:
    def reload(self,..):
        data = load_from_disk()
        some_dict = build_from_data(data)
        self.some_dict = some_dict

1. FastAPI后台任务的运行方式 #

FastAPI的后台任务运行在触发请求的同一个Worker进程的线程中，并非单独的进程。如果启动了多个Worker（比如用uvicorn --workers N），每个Worker都会有自己的后台任务实例，相互独立。

2. 字典重载的实现方案 #

核心原理 #

每个Worker进程拥有独立的内存空间，因此要让所有Worker同步更新字典，需要让每个Worker单独执行重载逻辑。以下两种方案均符合你允许Worker短暂不同步的需求：

方式一：定时自动重载（最简推荐） #

利用asyncio实现每个Worker内部定时执行重载逻辑，无需额外进程间通信：

from fastapi import APIRouter, FastAPI
from typing import Dict
import asyncio
import time

def load_data(path: str) -> Dict:
    # 模拟从磁盘加载数据逻辑
    return {"data": f"loaded at {time.time()}"}

def build_from_data(data: Dict) -> Dict:
    # 模拟构建复杂字典的逻辑
    return data.copy()

class Server:
    def __init__(self, some_dict: Dict[str, ...], reload_interval: int = 60):
        self.router = APIRouter()
        self.router.add_api_route("/bar", self.bar, methods=["POST"])
        self.some_dict = some_dict
        self.reload_interval = reload_interval
        # 启动定时重载后台任务
        asyncio.create_task(self._auto_reload())

    async def bar(self, text: str):
        # 使用当前最新的some_dict
        return {"text": text, "current_data": self.some_dict}

    async def _auto_reload(self):
        while True:
            await asyncio.sleep(self.reload_interval)
            try:
                raw_data = load_data("your_file_path")
                new_dict = build_from_data(raw_data)
                # 原子替换字典引用，线程安全（Python赋值操作是内存地址替换，不会被打断）
                self.some_dict = new_dict
                print(f"Worker {id(self)} completed reload at {time.time()}")
            except Exception as e:
                # 捕获重载异常，避免Worker进程崩溃
                print(f"Reload failed: {str(e)}")

# 启动代码
initial_dict = load_data("your_file_path")
server = Server(initial_dict, reload_interval=30)  # 每30秒重载一次
app = FastAPI()
app.include_router(server.router)

# 运行命令示例：uvicorn main:app --workers 4

方式二：API触发手动重载 #

如果需要手动触发重载而非定时，可以通过创建标记文件，让每个Worker定期检查并执行重载：

import os

# 在Server类中新增触发重载的API
@server.router.post("/trigger-reload")
async def trigger_reload():
    # 创建重载标记文件
    with open("reload_flag.txt", "w") as f:
        f.write("1")
    return {"message": "Reload signal sent to all workers"}

# 修改_auto_reload逻辑为检查标记文件
async def _auto_reload(self):
    while True:
        await asyncio.sleep(5)  # 每5秒检查一次标记
        if os.path.exists("reload_flag.txt"):
            try:
                os.remove("reload_flag.txt")
                raw_data = load_data("your_file_path")
                new_dict = build_from_data(raw_data)
                self.some_dict = new_dict
                print(f"Worker {id(self)} completed reload at {time.time()}")
            except Exception as e:
                print(f"Reload failed: {str(e)}")

关键注意事项 #

• 线程安全：替换字典引用的操作self.some_dict = new_dict是原子性的，不会出现半更新状态，多线程访问时无需额外锁。
• Worker独立性：每个Worker会独立执行重载逻辑，因此不同Worker的字典更新时间可能有差异，符合你允许短暂不同步的需求。
• 避免全局变量：不要使用全局字典，因为每个Worker的全局变量是独立的，无法跨Worker同步。
• 异常防护：重载逻辑必须加异常捕获，防止加载磁盘数据失败导致Worker进程崩溃。

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