Flask+Angular实时数据推送故障排查及替代方案咨询

问题描述 #

我通过Flask服务器为Angular前端提供REST API服务，现需添加服务器到客户端的实时推送机制处理实时数据。尝试使用Flask-SocketIO（Python端）和socket.io-client（Angular端）实现，但数据无法正常推送，且连接出现解析错误。以下是相关代码、日志及错误信息，请求排查问题或提供其他可行方案。

服务器代码 #

app = create_app()
socketioR.__init__(app, cors_allowed_origins="*", logger=True, engineio_logger=True, log_output=True)
manager = Manager(app)

manager.add_command('runserver', Server(host='localhost', port=settings.SYSTEM_PORT, threaded=True))

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

emit函数代码 #

def send_data():
    try:
        socketioR.send('message', {'data': 42}, namespace='/', broadcast=True)
    except Exception as e:
        print(e)

Angular客户端代码 #

import { Injectable } from '@angular/core';
import { BehaviorSubject, Observable } from 'rxjs';
import  io from "socket.io-client";
// import { Socket } from 'ngx-socket-io';


@Injectable({
  providedIn: 'root'
})
export class PushService {

    public message$: BehaviorSubject<any> = new BehaviorSubject('');
  private socket: any;
  constructor() {
    try {
      this.socket = io('http://localhost:8119', {
        extraHeaders: {
          'Access-Control-Allow-Origin': "*"
        }
      });
      this.socket.on('connect_error',(message: any)=>
        console.log(message))
    }
    catch (ex) {
      console.log(ex)
    }
    this.getNewMessage()
  }



  public getNewMessage(){
    this.socket.on('message', (message: any) =>{
      this.message$.next(message);
    });

    return this.message$.asObservable();
  };

}

服务器重复日志 #

INFO:werkzeug:127.0.0.1 - - [17/Jul/2022 11:45:22] "OPTIONS /socket.io/?EIO=4&transport=polling&t=O8BLMRO HTTP/1.1" 200 -
INFO:werkzeug:127.0.0.1 - - [17/Jul/2022 11:45:22] "GET /socket.io/?EIO=4&transport=polling&t=O8BLMRO HTTP/1.1" 200 -

客户端连接错误 #

Error: server error
    at Socket.onPacket (socket.js:317:1)
    at Emitter.emit (index.mjs:136:1)
    at Polling.onPacket (transport.js:105:1)
    at callback (polling.js:118:18)
    at Array.forEach (<anonymous>)
    at Polling.onData (polling.js:121:53)
    at Emitter.emit (index.mjs:136:1)
    at Request.onLoad (polling.js:365:18)
    at xhr.onreadystatechange [as __zone_symbol__ON_PROPERTYreadystatechange] (polling.js:301:26)
    at XMLHttpRequest.wrapFn (zone.js:766:1)

排查方案 #

• 修复服务器启动逻辑：manager.run()是阻塞调用，会启动Werkzeug原生服务器，后续的socketioR.run(app)根本不会执行。应移除Manager的runserver命令，直接用Flask-SocketIO启动服务器：

  if __name__ == '__main__':
      socketioR.run(app, host='localhost', port=settings.SYSTEM_PORT, debug=True, threaded=True)
  

  或者调整执行顺序，确保SocketIO服务器接管请求处理。

• 修正emit函数调用：socketioR.send()的参数格式错误，若要发送指定事件名的消息，需使用emit()方法。send()默认发送message事件，第一个参数是消息内容，而非事件名。修改后的代码：

  def send_data():
      try:
          socketioR.emit('message', {'data': 42}, namespace='/', broadcast=True)
      except Exception as e:
          print(e)
  

• 移除客户端无效请求头：Access-Control-Allow-Origin是服务器响应头，客户端请求中添加该头无意义，反而可能干扰连接。简化客户端连接代码：

  this.socket = io('http://localhost:8119');
  

• 检查版本兼容性：Flask-SocketIO与socket.io-client版本必须匹配，版本不兼容会导致协议解析错误。例如：

  • Flask-SocketIO 5.x 对应 socket.io-client 4.x
  • Flask-SocketIO 6.x 对应 socket.io-client 5.x
    分别执行以下命令检查版本：

  # Python端
  pip show flask-socketio
  # Angular端
  npm list socket.io-client
  

  若版本不匹配，升级或降级至兼容版本。

其他可行方案 #

• Server-Sent Events（SSE）：适合单向实时推送场景，Flask可直接实现SSE接口，Angular通过EventSourceAPI接收数据，无需额外依赖复杂的SocketIO库。
• Redis配合Flask-SocketIO：若后续需要多进程或分布式部署，可使用Redis作为消息队列实现跨进程推送，但需先解决当前单服务器的连接问题。

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