FastAPI单例依赖跨进程数据同步异常问题求助

你的问题核心在于多进程的内存隔离特性：基于线程锁实现的单例只能保证同一进程内的线程共享实例，但每个进程拥有独立的内存空间，启动新进程时会复制父进程的内存数据，因此不同进程的Database实例完全独立，修改互不影响。

解决方案1：使用多进程共享内存（multiprocessing.Manager） #

Python的multiprocessing.Manager提供了跨进程共享的数据结构，能让多个进程访问同一数据对象。修改你的单例类如下：

import threading
from multiprocessing import Manager

class Database:
    _instance = None
    _lock = threading.Lock()

    def __new__(cls, *args, **kwargs):
        if not cls._instance:
            with cls._lock:
                if not cls._instance:
                    cls._instance = super().__new__(cls)
                    # 通过Manager创建跨进程共享的命名空间
                    manager = Manager()
                    cls._instance.shared_data = manager.Namespace()
                    cls._instance.shared_data.volume = 0
        return cls._instance

    def replace_volume(self, value: int):
        # 直接修改共享数据，无需额外锁（Manager内部已处理同步）
        self.shared_data.volume = value

    def get_volume(self):
        return self.shared_data.volume

解决方案2：使用外部缓存（如Redis） #

将共享数据存储到外部缓存服务，所有进程统一读写缓存，天然支持跨进程甚至跨机器同步。示例代码：

import threading
import redis

class Database:
    _instance = None
    _lock = threading.Lock()

    def __new__(cls, *args, **kwargs):
        if not cls._instance:
            with cls._lock:
                if not cls._instance:
                    cls._instance = super().__new__(cls)
                    # 初始化Redis客户端
                    cls._instance.redis_client = redis.Redis(host='localhost', port=6379, db=0)
                    # 设置默认值（如果不存在）
                    if not cls._instance.redis_client.exists("volume"):
                        cls._instance.redis_client.set("volume", 0)
        return cls._instance

    def replace_volume(self, value: int):
        self.redis_client.set("volume", value)

    def get_volume(self):
        return int(self.redis_client.get("volume"))

解决方案3：使用消息队列同步变更 #

通过消息队列（如Redis Pub/Sub）在进程间推送数据变更，当一个进程修改数据时，发布变更消息，其他进程订阅并更新本地状态（或直接从消息获取最新值）。示例思路：

1. 修改数据时发布消息：

def replace_volume(self, value: int):
    self.redis_client.set("volume", value)
    # 发布变更消息
    self.redis_client.publish("volume_updates", str(value))

2. 在FastAPI的WebSocket路由中订阅消息：

@app.websocket("/stream/volume")
async def websocket_volume(websocket: WebSocket):
    await websocket.accept()
    # 创建Redis订阅客户端（需单独实例，避免阻塞）
    pubsub = db.redis_client.pubsub()
    pubsub.subscribe("volume_updates")
    
    try:
        # 初始发送当前值
        await websocket.send_text(json.dumps({"volume": db.get_volume()}))
        # 监听消息
        while True:
            message = pubsub.get_message(ignore_subscribe_messages=True)
            if message:
                volume = int(message['data'])
                await websocket.send_text(json.dumps({"volume": volume}))
            await asyncio.sleep(0.1)
    except Exception:
        print("WebSocket disconnected")
        pubsub.unsubscribe()
        await websocket.close()

总结 #

• 线程锁实现的单例仅作用于同一进程内的线程，多进程场景下必须依赖IPC机制或外部共享存储。
• 轻量场景优先用multiprocessing.Manager，分布式或跨机器场景优先用Redis等外部缓存。

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