Python Flask跨WSGI worker共享计数器变量失效问题咨询

进程内全局变量、基于进程内存的缓存（比如Flask-Caching默认的SimpleCache、ox_cache默认存储）的作用域都局限在单个Worker进程内部，不同Worker之间数据完全隔离，这就是本地单进程运行正常、多Worker部署后计数失效的核心原因。如果缓存用了进程内存后端，哪怕读写逻辑写得再正确，也不可能实现跨Worker共享计数。

要实现全局累计计数，必须满足两个前提：

• 计数数据存储在所有Worker都能访问的独立介质中，不依赖单个进程内存
• 计数的自增、阈值判断、重置操作必须保证原子性，避免多Worker并发请求时出现竞态，导致计数不准、函数重复触发

以下是两种生产可用的实现方案：

方案1：Redis实现（生产环境首选） #

Redis是独立部署的内存存储服务，所有Worker都可以通过网络连接访问，原生支持原子自增操作，不需要自行实现锁逻辑，性能高、稳定性强，是这类场景的首选方案。
先安装依赖：
pip install flask redis

实现代码：

from flask import Flask, render_template
import redis

app = Flask(__name__)
# 初始化Redis连接，所有Worker连接到同一个Redis实例
redis_client = redis.Redis(host='127.0.0.1', port=6379, db=0, decode_responses=True)
COUNTER_KEY = "global_request_counter"
# 初始化计数器，不存在时设为0
redis_client.setnx(COUNTER_KEY, 0)

def second_function():
    print('10 operations done')

@app.route('/')
def index():
    # 原子自增，直接返回自增后的值，Redis侧单命令执行无竞态
    current_count = redis_client.incr(COUNTER_KEY)
    if current_count >= 10:
        second_function()
        # 触发后重置计数器
        redis_client.set(COUNTER_KEY, 0)
    return render_template('index.html')

如果Redis和Flask服务部署在不同服务器，替换host参数为Redis实际服务地址即可，只要所有Worker能连通该Redis实例，计数就会全局生效。

方案2：文件锁+本地文件实现（无额外服务依赖） #

如果不想额外部署Redis这类服务，在Linux/Unix服务器上可以用固定磁盘文件存储计数，配合系统级文件锁保证多进程读写互斥，适合访问量不高的轻量场景。
实现代码：

from flask import Flask, render_template
import fcntl
import os

app = Flask(__name__)
COUNTER_FILE_PATH = "/tmp/flask_global_counter"
# 服务启动时初始化计数文件
if not os.path.exists(COUNTER_FILE_PATH):
    with open(COUNTER_FILE_PATH, "w") as f:
        f.write("0")

def second_function():
    print('10 operations done')

@app.route('/')
def index():
    current_count = 0
    # 加排他锁操作文件，保证同一时间只有一个Worker能读写计数
    with open(COUNTER_FILE_PATH, "r+") as f:
        fcntl.flock(f.fileno(), fcntl.LOCK_EX)
        try:
            current_count = int(f.read().strip())
            current_count += 1
            if current_count >= 10:
                second_function()
                current_count = 0
            # 写回新的计数值
            f.seek(0)
            f.truncate()
            f.write(str(current_count))
        finally:
            # 释放文件锁
            fcntl.flock(f.fileno(), fcntl.LOCK_UN)
    return render_template('index.html')

注意：Windows系统不支持fcntl模块，该方案仅可在Linux、MacOS等类Unix系统上使用。

之前Flask-Caching方案失效的排查点 #

如果要继续用Flask-Caching实现，必须把缓存后端换成Redis、Memcached这类独立于Worker进程的外部缓存服务，绝对不能用默认的SimpleCache内存后端。同时计数自增要调用缓存客户端自带的原子incr方法，不要自行实现「先get值→加1→set值」的逻辑，否则并发场景下依然会出现计数偏差。

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