Flask Web应用无需数据库存储全局动态变量的方案咨询

嘿，这个场景我太有共鸣了——就3个全局动态变量，犯不上搭个重型数据库，但文本/pickle那套又容易出竞态问题，给你几个实用的替代方案，按需选就行：

方案1：带文件锁的JSON存储 #

这是对原有文件存储方案的升级，通过文件锁解决竞态问题，不用额外依赖服务，数据还能持久化（重启不丢），JSON格式也比pickle更可读安全。

优点 #

• 零额外服务依赖，上手快
• 数据持久化，重启应用不会丢失
• JSON格式易读，方便手动修改（如果需要）

缺点 #

• 并发高时性能不如缓存/Redis，适合中小流量场景

代码示例 #

先安装依赖：pip install filelock

from filelock import FileLock
import json
import os
from flask import Flask

app = Flask(__name__)

# 定义存储文件和锁文件路径
DATA_FILE = "global_vars.json"
LOCK_FILE = "global_vars.lock"

def get_global_vars():
    # 加锁后再读写，避免多进程/线程同时操作
    with FileLock(LOCK_FILE):
        if not os.path.exists(DATA_FILE):
            # 初始化默认值
            return {"var1": "", "var2": 0, "var3": False}
        with open(DATA_FILE, "r", encoding="utf-8") as f:
            return json.load(f)

def update_global_vars(new_vars):
    with FileLock(LOCK_FILE):
        current_vars = get_global_vars()
        current_vars.update(new_vars)
        with open(DATA_FILE, "w", encoding="utf-8") as f:
            json.dump(current_vars, f, indent=2)

# 示例接口
@app.route("/get-vars")
def get_vars():
    return get_global_vars()

@app.route("/update-var/<var_name>/<value>")
def update_var(var_name, value):
    # 这里可以根据变量类型做转换，比如把value转成int/float
    if var_name == "var2":
        value = int(value)
    elif var_name == "var3":
        value = value.lower() == "true"
    update_global_vars({var_name: value})
    return f"Updated {var_name} to {value}"

if __name__ == "__main__":
    app.run(debug=True)

方案2：Flask-Caching内存缓存 #

如果你的应用是单进程部署（比如用Flask自带的开发服务器，或者Gunicorn单worker），内存缓存是个极快的选择，而且Flask-Caching自带锁机制解决竞态问题。

优点 #

• 读写速度极快，适合高频访问
• 代码简洁，和Flask集成度高
• 自带锁机制，避免并发更新冲突

缺点 #

• 多进程部署时，每个worker有独立的缓存空间，变量无法同步
• 重启应用会丢失所有缓存数据，适合可以接受临时数据的场景

代码示例 #

先安装依赖：pip install flask-caching

from flask import Flask
from flask_caching import Cache

app = Flask(__name__)

# 配置内存缓存，timeout设为0表示永不过期
cache = Cache(app, config={
    "CACHE_TYPE": "SimpleCache",
    "CACHE_DEFAULT_TIMEOUT": 0
})

# 初始化全局变量
if not cache.get("global_vars"):
    cache.set("global_vars", {"var1": "", "var2": 0, "var3": False})

@app.route("/get-vars")
def get_vars():
    return cache.get("global_vars")

@app.route("/update-var/<var_name>/<value>")
def update_var(var_name, value):
    # 用缓存锁确保原子更新
    with cache.lock("global_vars_lock"):
        current_vars = cache.get("global_vars")
        # 类型转换逻辑
        if var_name == "var2":
            value = int(value)
        elif var_name == "var3":
            value = value.lower() == "true"
        current_vars[var_name] = value
        cache.set("global_vars", current_vars)
    return f"Updated {var_name} to {value}"

if __name__ == "__main__":
    app.run(debug=True)

方案3：Redis轻量级存储 #

如果需要支持多进程/多服务器共享变量，或者并发量较高，Redis是最优解——它是轻量级键值数据库，所有操作都是原子性的，完全避免竞态问题，还能持久化数据。

优点 #

• 原子操作，彻底解决竞态条件
• 支持多进程、多服务器共享数据
• 数据可持久化，重启Redis也不会丢失
• 性能极佳，适合高并发场景

缺点 #

• 需要安装Redis服务（不过非常轻量，用Docker一键就能跑：docker run -d -p 6379:6379 redis）
• 多了一个外部依赖，但维护成本极低

代码示例 #

先安装依赖：pip install redis

from flask import Flask
import redis
import json

app = Flask(__name__)

# 连接本地Redis
r = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)

# 初始化全局变量
if not r.exists("global_vars"):
    r.set("global_vars", json.dumps({"var1": "", "var2": 0, "var3": False}))

@app.route("/get-vars")
def get_vars():
    vars_json = r.get("global_vars")
    return json.loads(vars_json)

@app.route("/update-var/<var_name>/<value>")
def update_var(var_name, value):
    # 用Redis的Watch+Pipeline实现原子更新，避免并发冲突
    with r.pipeline() as pipe:
        while True:
            try:
                # 监听key，防止其他客户端修改
                pipe.watch("global_vars")
                current_vars = json.loads(pipe.get("global_vars"))
                # 类型转换
                if var_name == "var2":
                    value = int(value)
                elif var_name == "var3":
                    value = value.lower() == "true"
                current_vars[var_name] = value
                # 开启事务，执行更新
                pipe.multi()
                pipe.set("global_vars", json.dumps(current_vars))
                pipe.execute()
                break
            except redis.WatchError:
                # 如果key被修改，重试
                continue
    return f"Updated {var_name} to {value}"

if __name__ == "__main__":
    app.run(debug=True)

最后选择建议 #

• 中小流量、单进程、不想加额外服务：选带文件锁的JSON存储
• 单进程、高频访问、可接受数据临时丢失：选Flask-Caching内存缓存
• 多进程/多服务器、高并发、需要数据持久化：选Redis

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