python - 如何使用 python flask socket.io 和线程从 redis 异步获取实时数据？

Question

我的程序从传感器收集数据并将这些数据发送到 Web 界面，该界面以实时图形显示并发回一些命令。

问题是数据发送有很大的延迟或发送旧数据。我应该更改什么以向两种方式异步发送数据？

我简化了我的代码。示例1，当数据在几分钟内来一次时:

# System stuff
import os
import sys
import serial
# Multiprocessing
from multiprocessing import Process
from threading import Thread # for same memory
import timeit
from time import sleep
# database
import redis
# Web server
from flask import Flask, render_template
from flask_socketio import SocketIO, emit, send, join_room, leave_room, close_room, rooms, disconnect

# Config
DEBUG = True


db = redis.StrictRedis(host='localhost', port=6379, db=0)
#db vars
db.set('velocity', 0.0)
db.set('distance', 0.0)


def webServer(db):
    app = Flask(__name__)
    app.config['SECRET_KEY'] = 'secret!'
    socketio = SocketIO(app)


    @app.route('/')
    def index():
        return render_template('index.html')


    def exchangeData():
        prev_distance = 0.00
        while(True):
            distance = round(float(db.get('distance')), 2)
            if(distance != prev_distance):
                velocity = round(float(db.get('velocity')), 2)
                distance = round(float(db.get('distance')), 2)              
                socketio.emit('exchangeData', {
                    'velocity': velocity,
                    'distance': distance,
                })              
                prev_distance = distance                
                print("DATA sended: %s m/s" % velocity) 

            sleep(0.2)      

    @socketio.on('connect')
    def test_connect():
        t_exchangeData = Thread(target=exchangeData).start()    

    socketio.run(app, debug=True, host="0.0.0.0")


def newData(db):
    c = 0.00
    while(True):
        db.set('velocity', c)
        db.set('distance', c)
        c += 1.00
        sleep(1)




if __name__ == '__main__':
    p_newData = Process(target=newData, args=(db,)).start()
    p_webServer = Process(target=webServer, args=(db,)).start()
    #p_checkConnection = Process(target=checkConnection, args=(db, HOSTNAME, pinglog)).start()
    #p_calcVD = Process(target=calcVD, args=(db,)).start()

在浏览器中，我得到这个延迟大约 2 分钟:

Object {velocity: 218, distance: 218}
// two minutes later
Object {velocity: 306, distance: 306}
// two minutes later
Object {velocity: 306, distance: 306}

例子2，当我不使用if语句和sleep时:

# System stuff
import os
import sys
import serial
# Multiprocessing
from multiprocessing import Process
from threading import Thread # for same memory
import timeit
from time import sleep
# database
import redis
# Web server
from flask import Flask, render_template
from flask_socketio import SocketIO, emit, send, join_room, leave_room, close_room, rooms, disconnect

# Config
DEBUG = True


db = redis.StrictRedis(host='localhost', port=6379, db=0)
#db vars
db.set('velocity', 0.0)
db.set('distance', 0.0)


def webServer(db):
    app = Flask(__name__)
    app.config['SECRET_KEY'] = 'secret!'
    socketio = SocketIO(app)


    @app.route('/')
    def index():
        return render_template('index.html')


    def exchangeData():
        prev_distance = 0.00
        while(True):
            #distance = round(float(db.get('distance')), 2)
            #if(distance != prev_distance):
            velocity = round(float(db.get('velocity')), 2)
            distance = round(float(db.get('distance')), 2)              
            socketio.emit('exchangeData', {
                'velocity': velocity,
                'distance': distance,
            })              
            prev_distance = distance                
            print("DATA sended: %s m/s" % velocity) 

            #sleep(0.2)      

    @socketio.on('connect')
    def test_connect():
        t_exchangeData = Thread(target=exchangeData).start()    

    socketio.run(app, debug=True, host="0.0.0.0")


def newData(db):
    c = 0.00
    while(True):
        db.set('velocity', c)
        db.set('distance', c)
        c += 1.00
        sleep(1)




if __name__ == '__main__':
    p_newData = Process(target=newData, args=(db,)).start()
    p_webServer = Process(target=webServer, args=(db,)).start()
    #p_checkConnection = Process(target=checkConnection, args=(db, HOSTNAME, pinglog)).start()
    #p_calcVD = Process(target=calcVD, args=(db,)).start()

在这种情况下，我实时获取数据，但它是相同的数据，并且在几分钟内仅更改一次:

Object {distance: 3, velocity: 3}
Object {distance: 3, velocity: 3}
Object {distance: 3, velocity: 3}
// repeating around 2 minutes
Object {distance: 357, velocity: 357} 
// repeating again...

Answer 1

主要问题是我没有添加这两行:

import eventlet
eventlet.monkey_patch()

此外，当我试图找到解决方案时，当我在 while 循环之外从 redis 调用数据时犯了一个错误。