javascript - 通过 Websockets 从 Python Flask 服务器连续向客户端发送数据
问题描述
我正在设置一个仪表板,它需要 Python Flask 服务器每隔几秒左右获取一次公共交通数据,并通过 websockets 将此数据发送到客户端网页。我正在使用 FlaskSocketIO 来实现这一点。
例如,我可以在事件发生时发送消息'connect'
,但我一直无法找到持续更新此消息的方法。这是我的方法:
from flask import Flask, render_template, url_for, request
from flask_socketio import SocketIO, emit, send
import eventlet
eventlet.monkey_patch()
app = Flask(__name__)
socketio = SocketIO(app, logger=True, engineio_logger=True)
def listen():
while True:
message = # some updating message
emit('message', message)
socketio.sleep(1)
@app.route('/')
def index():
return render_template('index.html')
@socketio.on('connect')
def handle_connect():
listen()
if __name__ == "__main__":
socketio.run(app, debug=True)
记录器说它发送消息。但是,在客户端,没有收到任何消息。客户端 JavaScript 是:
var socket = io();
socket.on('message', function(msg) {
console.log(msg);
});
解决方案
我和你有同样的问题,在我的情况下,我想更新传感器仪表板,所以我研究了观察者模式,这个链接真的很好观察者模式。实现一个 SocketEmitObservator,使用你不需要创建监听器的观察者(无限时循环)并使您的代码更干净:
class SensorObserver(Observer):
sensor_id: int
sid = ''
def __init__(self, sid,sensor_id=None):
Observer.__init__(self)
self.sensor_id = sensor_id
self.sid = sid
SensorPublisher().attach(self)
def update(self, sensorData: DataModel) -> None:
socket.emit("info",
{'sensor': subject},
namespace='/api/sector/machine',
room=self.sid)
def __del__(self):
SensorPublisher().detach(self)
class SensorPublisher(Subject, Singleton):
_observers: List[SensorObserver] = []
_sensorData: Dict[str, DataModel] = {}
def attach(self, observer: SensorObserver) -> None:
if observer not in self._observers:
self._observers.append(observer)
def notify(self, sensorModel: DataModel) -> None:
for observer in self._observers:
if sensorModel.sensor_id == observer.sensor_id:
observer.update(sensorModel)
def detach(self, observer: SensorObserver) -> None:
if observer in self._observers:
ob = self._observers.pop(self._observers.index(observer))
del ob
def update(self, sensorModel: DataModel):
sendorId: str = str(sensorModel.sensor_id)
self._sensorData[sendorId] = sensorModel
self.notify(sensorModel)
#### And in the place where i update sensor data i put
SensorPublisher().update(dataStructList)
#### To create a observr just do
@socketio.on('connect')
def handle_connect():
SensorObserver(self.sid, item)
在我的情况下,我这样使用,但你可以适应你的问题
推荐阅读
- websocket - Twilio 媒体流 - Websocket 握手错误 31920
- memory-management - 在编译时如何实现引用计数器?
- python - Pygsheets:如何根据条件突出显示特定列中的单元格
- kotlin - 执行顺序 - Kotlin
- android - 将多个文件(主要是图像)上传到 Amazon S3 存储桶的最佳实践
- c - linux list_for_each_entry() 宏的正确参数是什么?
- python - 如何根据 Python 中数组的长度创建变量?
- python - 我的网络刮板没有刮掉所有的评论和用户名
- assembly - 如何在 x86 中进行反向偏移/索引
- gmail - 在 Gmail 中打开 amp 电子邮件时,放大器选择器不起作用,复选框部分完全关闭