使用Flask与Gunicorn时，如何在多Worker间共享Python字典？

核心原因 #

你之前的问题出在每个Worker进程单独初始化了multiprocessing.Manager，导致每个Worker持有独立的共享对象实例。Gunicorn的Worker是通过fork主进程创建的，必须在主进程（Worker启动前）初始化共享对象，才能让所有Worker继承同一个代理实例。

实现方案 #

1. 编写Gunicorn配置文件（gunicorn_config.py） #

在主进程启动阶段创建共享字典，再通过钩子传递给所有Worker：

from multiprocessing import Manager

# 主进程中保存全局共享对象
shared_dict = None
shared_lock = None

def on_starting(server):
    """Gunicorn启动前钩子：在主进程初始化共享资源"""
    global shared_dict, shared_lock
    manager = Manager()
    shared_dict = manager.dict()
    shared_lock = manager.Lock()  # 用于保证并发修改的安全性
    # 可选：预先初始化共享数据
    shared_dict["init_key"] = "init_value"

def post_worker_init(worker):
    """Worker初始化后钩子：将共享资源挂载到Flask应用"""
    from app import app
    app.config["SHARED_DICT"] = shared_dict
    app.config["SHARED_LOCK"] = shared_lock

2. 在Flask应用中使用共享字典（app.py） #

直接通过Flask配置获取共享对象，修改时记得加锁：

from flask import Flask, request

app = Flask(__name__)

@app.route("/set")
def set_shared_value():
    key = request.args.get("key")
    value = request.args.get("value")
    if not key or not value:
        return "Missing key or value"
    
    # 加锁避免并发修改冲突
    with app.config["SHARED_LOCK"]:
        app.config["SHARED_DICT"][key] = value
    return f"Successfully set {key} = {value}"

@app.route("/get")
def get_shared_value():
    key = request.args.get("key")
    if not key:
        return "Missing key"
    return app.config["SHARED_DICT"].get(key, "Key not found")

3. 启动Gunicorn #

指定配置文件启动4个Worker：

gunicorn -w 4 -b 0.0.0.0:5000 --config gunicorn_config.py app:app

关键注意点 #

• 性能限制：multiprocessing.Manager基于进程间通信（IPC）实现，高并发场景下会有性能损耗，仅适合低并发的共享需求
• 数据持久化：如果Gunicorn主进程重启，共享字典的数据会丢失，若需要持久化仍需依赖数据库或缓存工具
• 锁的必要性：多个Worker同时修改共享字典时，必须加锁保证数据一致性，否则会出现竞态条件

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