Flask-SocketIO：Worker脚本发送的消息无法被前端接收的问题

核心问题是：独立运行的Worker脚本中的SocketIO实例和Flask主应用的SocketIO实例没有共享通信通道，导致Worker发送的事件无法传递到客户端。解决方法是通过消息代理实现两者的通信，让Worker的事件能通过代理转发给主应用，再由主应用推送给客户端。

步骤1：安装依赖 #

以Redis作为消息代理为例，安装所需包：

pip install flask-socketio redis

步骤2：配置Flask主应用的SocketIO #

修改main.py，初始化SocketIO时指定消息代理地址：

from flask import Flask
from flask_socketio import SocketIO
from flask_login import login_required
from flask_breadcrumbs import register_breadcrumb

app = Flask(__name__)
# 配置Redis作为消息代理，确保Redis服务运行在对应地址
socketio = SocketIO(app, message_queue='redis://localhost:6379/0')

@app.route("/home", methods=['POST', 'GET'])
@login_required
@register_breadcrumb(app, '.', 'Home')
def home():
    print("Emitting test messages")
    socketio.emit('edit_plan', {'data': 'testing test'})
    return "Home Page"

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

步骤3：修改Worker脚本连接到消息代理 #

修改worker.py，不需要初始化Flask app，直接创建连接到同一消息代理的SocketIO实例：

from flask_socketio import SocketIO
import time

# 连接到同一个Redis消息代理
socketio = SocketIO(message_queue='redis://localhost:6379/0')

def long_running_task():
    # 模拟耗时任务
    print("Running long task...")
    time.sleep(5)
    # 发送消息到客户端
    socketio.emit('edit_plan', {'data': 'testing test from worker'})

if __name__ == '__main__':
    long_running_task()

原理说明 #

消息代理（如Redis）作为中间通信层：

• Flask主应用的SocketIO实例和Worker的SocketIO实例都连接到同一个代理
• Worker调用socketio.emit()时，事件会被发送到消息代理
• 主应用的SocketIO实例监听代理中的事件，收到后立即推送给所有连接的客户端

注意事项 #

• 确保Redis服务已启动并正常运行（可通过redis-cli ping验证）
• 若使用RabbitMQ作为代理，只需将message_queue改为amqp://localhost
• 生产环境部署Flask应用时，需使用支持WebSocket的Worker，例如Gunicorn搭配geventwebsocket：

  gunicorn -k geventwebsocket.gunicorn.workers.GeventWebSocketWorker main:app
  

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