Flask状态路由无法正常更新，如何实现前端动态状态推送？

问题根源 #

你用全局变量current_status的方式在**多进程/多线程环境下（比如生产部署的Flask）**会失效，因为每个进程/线程都有自己的全局变量副本，更新的状态无法跨进程共享。即使是调试模式，Flask默认会启动两个进程（主进程+重载进程），也可能导致状态不同步。

解决方案一：单进程调试环境下修复全局变量 #

如果只是本地调试用，可以强制Flask以单进程启动，确保全局变量能共享：

修改启动代码：

if __name__ == '__main__':
    app.run(debug=True, threaded=False, processes=1)

这种设置能让全局变量的更新在/status路由中正确返回，但仅适合本地测试，绝对不能用于生产环境。

解决方案二：生产环境用共享存储（推荐） #

生产环境必须用跨进程共享的存储来保存状态，比如Redis、Memcached或数据库。这里以Redis为例：

步骤1：安装依赖 #

pip install redis

步骤2：修改代码 #

from flask import Flask, jsonify
import redis

app = Flask(__name__)
# 连接本地Redis服务（默认端口6379）
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

# 初始化初始状态
if not r.get('current_status'):
    r.set('current_status', 'Initial Status')

@app.route('/status', methods=['GET'])
def get_status():
    return jsonify({'status': r.get('current_status')})

@app.route('/update_status/<string:new_status>', methods=['PUT'])
def update_status(new_status):
    r.set('current_status', new_status)
    return jsonify({'message': 'Status has been updated', 'new_status': new_status})

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

这种方式下，无论多少进程都能读取到最新的共享状态，完全适配生产环境。

解决方案三：主动推送状态（更优用户体验） #

前端轮询/status路由效率较低，推荐用Server-Sent Events (SSE) 实现服务器主动推送状态更新，前端无需反复请求。

后端代码（Flask SSE实现） #

from flask import Flask, jsonify, Response, stream_with_context
import redis
import time

app = Flask(__name__)
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

# 初始化初始状态
if not r.get('current_status'):
    r.set('current_status', 'Initial Status')

@app.route('/status/stream')
def status_stream():
    def generate():
        last_status = None
        while True:
            current_status = r.get('current_status')
            # 仅在状态变化时推送
            if current_status != last_status:
                yield f"data: {current_status}\n\n"
                last_status = current_status
            time.sleep(1)  # 每秒检查一次状态变化
    return Response(stream_with_context(generate()), mimetype='text/event-stream')

@app.route('/update_status/<string:new_status>', methods=['PUT'])
def update_status(new_status):
    r.set('current_status', new_status)
    return jsonify({'message': 'Status has been updated', 'new_status': new_status})

if __name__ == '__main__':
    app.run(debug=True, threaded=True)

前端代码（监听SSE） #

const eventSource = new EventSource('/status/stream');
eventSource.onmessage = function(event) {
    console.log('当前状态:', event.data);
    // 更新页面上的状态显示元素
    document.getElementById('status-display').textContent = event.data;
};

// 连接错误处理
eventSource.onerror = function(error) {
    console.error('SSE连接出错:', error);
    eventSource.close();
};

这种方式前端能实时接收状态更新，无需轮询，用户体验更流畅。

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