前言
定制网站本人于年初基于+echarts+mysql+ajax定制网站实现了自动刷新,但由于ajax定制网站轮询的弊端(定制网站请求必须由客户端向服务端发起,定制网站然后服务端进行响应),想看ajax定制网站实现的朋友可以看我写的这篇。定制网站现改用双向传输、定制网站推送消息方面能够做到灵活、简便、定制网站高效实现方案,定制网站即数据库收到数据立刻定制网站向客户端发送数据,定制网站无需客户端先向数据库发送请求。
一、环境准备
定制网站网上已经有许多教程,但由于websocket定制网站版本匹配和引入库版本等问题,定制网站大部分截至我发文阶段定制网站都无法直接实现,定制网站经过本人测试更改,定制网站以下为实现代码,定制网站另附上我的相关版本,定制网站以免读者走弯路,定制网站如需要查看其他版本的匹配方案,见
以下为官方包版本匹配说明:
以下为本人操作案例版本:
二、Flask+websocket
服务端app.py:
from flask import Flask, render_templatefrom flask_socketio import SocketIO, emitfrom threading import Lockimport randomasync_mode = Noneapp = Flask(__name__)app.config['SECRET_KEY'] = 'secret!'socketio = SocketIO(app)thread = Nonethread_lock = Lock()@app.route('/')def index(): return render_template('test.html')@socketio.on('connect', namespace='/test_conn')def test_connect(): global thread with thread_lock: if thread is None: thread = socketio.start_background_task(target=background_thread)def background_thread(): while True: socketio.sleep(5) t = random.randint(1, 100) socketio.emit('server_response', {'data': t}, namespace='/test_conn')if __name__ == '__main__': socketio.run(app, debug=True)
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
客户端test.html:
注意以下引入版本,如果报错基本都是版本不匹配的问题。
<!DOCTYPE html><html lang="en"><head> <meta charset="UTF-8"> <title></title> <script type="text/javascript" src="//cdn.bootcss.com/jquery/3.1.1/jquery.min.js"></script> <script type="text/javascript" src="//cdn.bootcss.com/socket.io/1.5.1/socket.io.min.js"></script></head><body><h2 id="t"></h2><script type="text/javascript"> $(document).ready(function() { namespace = '/test_conn'; var socket = io.connect(location.protocol + '//' + document.domain + ':' + location.port + namespace); socket.on('server_response', function(res) { console.log(res.data) var t = res.data; $("#t").text(t); }); });</script></body></html>
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
实现结果:
附上
三、Flask+echarts+websocket
服务端:
# encoding:utf-8# !/usr/bin/env pythonimport psutilimport timefrom threading import Lockfrom flask import Flask, render_templatefrom flask_socketio import SocketIOasync_mode = Noneapp = Flask(__name__)app.config['SECRET_KEY'] = 'secret!'socketio = SocketIO(app, async_mode=async_mode)thread = Nonethread_lock = Lock()# 后台线程 产生数据,即刻推送至前端def background_thread(): count = 0 while True: socketio.sleep(5) count += 1 t = time.strftime('%M:%S', time.localtime()) # 获取系统时间(只取分:秒) cpus = psutil.cpu_percent(interval=None, percpu=True) # 获取系统cpu使用率 non-blocking socketio.emit('server_response', {'data': [t, cpus], 'count': count}, namespace='/test') # 注意:这里不需要客户端连接的上下文,默认 broadcast = True@app.route('/')def index(): return render_template('test.html', async_mode=socketio.async_mode)@socketio.on('connect', namespace='/test')def test_connect(): global thread with thread_lock: if thread is None: thread = socketio.start_background_task(target=background_thread)if __name__ == '__main__': socketio.run(app, debug=True)
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
客户端:
<!DOCTYPE html><html lang="en"><head> <meta charset="UTF-8"> <title></title> <script type="text/javascript" src="//cdn.bootcss.com/jquery/3.1.1/jquery.min.js"></script> <script type="text/javascript" src="//cdn.bootcss.com/socket.io/1.5.1/socket.io.min.js"></script> <!-- ECharts 引入 --> <script src="https://cdn.staticfile.org/echarts/4.3.0/echarts.min.js"></script></head><body><div id="main" style="height:500px;border:1px solid #ccc;padding:10px;"></div> <script type="text/javascript"> var myChart = echarts.init(document.getElementById('main')); myChart.setOption({ title: { text: '系统监控走势图' }, tooltip: {}, legend: { data:['cpu'] }, xAxis: { data: [] }, yAxis: {}, series: [{ name: 'cpu', type: 'line', data: [] }] }); var time = ["","","","","","","","","",""], cpu = [0,0,0,0,0,0,0,0,0,0] //准备好统一的 callback 函数 var update_mychart = function (res) { //res是json格式的response对象 // 隐藏加载动画 myChart.hideLoading(); // 准备数据 time.push(res.data[0]); cpu.push(parseFloat(res.data[1])); if (time.length >= 10){ time.shift(); cpu.shift(); } // 填入数据 myChart.setOption({ xAxis: { data: time }, series: [{ name: 'cpu', // 根据名字对应到相应的系列 data: cpu }] }); }; // 首次显示加载动画 myChart.showLoading(); // 建立socket连接,等待服务器“推送”数据,用回调函数更新图表 $(document).ready(function() { namespace = '/test'; var socket = io.connect(location.protocol + '//' + document.domain + ':' + location.port + namespace); socket.on('server_response', function(res) { update_mychart(res); }); }); </script></body></html>
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
效果:
见。
四、添加数据库测试
思路和第三节差不多,写一个线程不断往数据库中插入数据,客户端显示横轴为数据库插入时间,纵轴为相应数据。
本人参考这几篇文章、、。
-
创建数据库
这一步可以直接从数据库软件navicate、mysql命令行或者是flask框架的拓展包来进行,为了方便后续操作,在这里使用flask框架的数据库拓展包Flask-SQLAlchemy,创建测试类数据库代码如下:db_create.py
from flask import Flaskfrom flask_sqlalchemy import SQLAlchemyimport pymysqlpymysql.install_as_MySQLdb()app = Flask(__name__)class Config(object): """配置参数""" # 设置连接数据库的URL user = 'root' password = 自己数据库的密码 database = 'test' SQLALCHEMY_DATABASE_URI = 'mysql://%s:%s@127.0.0.1:3306/%s' % (user, password, database) # 设置sqlalchemy自动更跟踪数据库 SQLALCHEMY_TRACK_MODIFICATIONS = True # 查询时会显示原始SQL语句 SQLALCHEMY_ECHO = True # 禁止自动提交数据处理 SQLALCHEMY_COMMIT_ON_TEARDOWN = False# 读取配置app.config.from_object(Config)# 创建数据库sqlalchemy工具对象db = SQLAlchemy(app)class Test(db.Model): # 定义表名 __tablename__ = 'sea_data' # 定义字段 id = db.Column(db.Integer, primary_key=True, autoincrement=True) # id 主键、自增 record_t = db.Column(db.DateTime, unique=True) # record_t 上传时间 temperature = db.Column(db.Float) # 气温数值if __name__ == '__main__': # 删除所有表 db.drop_all() # 创建所有表 db.create_all()
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
执行该脚本
python db_create.py
,看到表创建完成
-
定时插入数据
参考,以下为对db_create.py的补充:def insert(): print("定时器启动了") print(threading.current_thread()) # 查看当前线程 record_t = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) temperature = round(random.uniform(0, 40), 2) # 产生一个0-40之间的数字,保留两位小数 print('hello') ins = Test(record_t=record_t, temperature=temperature) db.session.add(ins) db.session.commit() print('插入成功!') timer = threading.Timer(5, insert) # 在run函数结束之前我再开启一个定时器 timer.start()if __name__ == '__main__': # 删除所有表 db.drop_all() # 创建所有表 db.create_all() t1 = threading.Timer(5, function=insert) # 过5s之后我执行后面的一个函数,开启一个线程 t1.start() # 设置一个多线程 # while True: # time.sleep(10) # 延时10s # print('主线程')
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
运行
python db_create.py
结果,5秒插入一次随机数据:
-
查询数据
想到两种方式,一种是通过查询数据库取最近十条记录传给前端渲染显示;另一种是只查询最近的一条,通过前端将最前一条数据挤出去显示。
在这里我们使用第二种方法,就查询最近1条显示,前端会不断地将以往的数据挤出去并填充。def query(): # 查询最近一条数据 # 只有最后加.all()才能读到实例,order_by和limit是条件查询 new = db.session.query(Test).order_by(Test.id.desc()).limit(1).all() print(new) print( class_to_dict(new)) # [{'temperature': 23.18, 'id': 5, 'record_t': datetime.datetime(2022, 10, 8, 10, 41, 35)}]# 查询结果转为字典def class_to_dict(obj): is_list = obj.__class__ == [].__class__ is_set = obj.__class__ == set().__class__ if is_list or is_set: obj_arr = [] for o in obj: dict = {} a = o.__dict__ if "_sa_instance_state" in a: del a['_sa_instance_state'] dict.update(a) obj_arr.append(dict) return obj_arr else: dict = {} a = obj.__dict__ if "_sa_instance_state" in a: del a['_sa_instance_state'] dict.update(a) return dict
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
-
websocket查询数据
在从数据库获取最新数据的时候出现了一个问题,插入的数据无法被获取,收到的数据都很旧,经过查询资料,这篇文章解释了为什么,需要先清空缓存,再查询:#清空缓存 db_session.commit()
- 1
- 2
在这里实现源码,注释我都写在源码里了,在这里不过多做赘述,需要的朋友直接copy,这是demo结构:
后端 app_dbchart.py:
# encoding:utf-8# !/usr/bin/env pythonimport psutilimport timefrom threading import Lockfrom flask import Flask, render_templatefrom flask_socketio import SocketIOimport threadingfrom db_create import insert, queryasync_mode = Noneapp = Flask(__name__)app.config['SECRET_KEY'] = 'secret!'socketio = SocketIO(app, async_mode=async_mode)thread = Nonethread_lock = Lock()# 后台线程 产生数据,即刻推送至前端def background_thread(): count = 0 while True: socketio.sleep(5) count += 1 # 目前问题,数据库无法读取到最新数据 query方法 print(query()) temperature = query()['temperature'] record_t = query()['record_t'] # t = time.strftime('%Y%M:%S', time.localtime()) # # 获取系统时间(只取分:秒) # cpus = psutil.cpu_percent(interval=None, percpu=True) # # 获取系统cpu使用率 non-blocking socketio.emit('server_response', {'data': [record_t, temperature], 'count': count}, namespace='/test') # 注意:这里不需要客户端连接的上下文,默认 broadcast = True@app.route('/')def index(): return render_template('test_dbchart.html', async_mode=socketio.async_mode)@socketio.on('connect', namespace='/test')def test_connect(): global thread with thread_lock: if thread is None: thread = socketio.start_background_task(target=background_thread)if __name__ == '__main__': # 定时插入后来个定时画图 socketio.run(app, debug=True)
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
db_create.py:
import jsonimport randomimport timefrom flask import Flaskfrom flask_sqlalchemy import SQLAlchemyimport pymysqlimport datetimeimport threadingpymysql.install_as_MySQLdb()app = Flask(__name__)class Config(object): """配置参数""" # 设置连接数据库的URL user = 'root' password = 自己的密码 database = 'test' SQLALCHEMY_DATABASE_URI = 'mysql://%s:%s@127.0.0.1:3306/%s' % (user, password, database) # 设置sqlalchemy自动更跟踪数据库 SQLALCHEMY_TRACK_MODIFICATIONS = True # 查询时会显示原始SQL语句 SQLALCHEMY_ECHO = True # 禁止自动提交数据处理 SQLALCHEMY_COMMIT_ON_TEARDOWN = False ENV = 'development' DEBUG = True# 读取配置app.config.from_object(Config)# 创建数据库sqlalchemy工具对象db = SQLAlchemy(app)class Test(db.Model): # 定义表名 __tablename__ = 'sea_data' # 定义字段 id = db.Column(db.Integer, primary_key=True, autoincrement=True) # id 主键、自增 record_t = db.Column(db.DateTime, unique=True) # record_t 上传时间 temperature = db.Column(db.Float) # 气温数值def insert(): print("定时器启动了") print(threading.current_thread()) # 查看当前线程 record_t = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) temperature = round(random.uniform(0, 40), 2) # 产生一个0-40之间的数字,保留两位小数 ins = Test(record_t=record_t, temperature=temperature) db.session.add(ins) db.session.commit() print('插入成功!') timer = threading.Timer(5, insert) # 在insert函数结束之前我再开启一个定时器 timer.start()def create(): # 创建所有表 db.create_all()def drop(): # 删除所有表 db.drop_all()def query(): # 清空缓存 db.session.commit() # 查询最近一条数据 # 只有最后加.all()才能读到实例,order_by和limit是条件查询 new = db.session.query(Test).order_by(Test.id.desc()).limit(1).all() # [{'temperature': 23.18, 'id': 5, 'record_t': datetime.datetime(2022, 10, 8, 10, 41, 35)}] list result = class_to_dict(new) # 取的时间json.dumps无法对字典中的datetime时间格式数据进行转化。因此需要添加特殊日期格式转化 result[0]['record_t'] = json.dumps(result[0]['record_t'], cls=DateEncoder) # print(result[0]) # {'temperature': 23.18, 'id': 5, 'record_t': '"2022-10-08 10:41:35"'} return result[0] # {'temperature': 23.18, 'id': 5, 'record_t': '"2022-10-08 10:41:35"'} # timer = threading.Timer(5, query) # 在insert函数结束之前我再开启一个定时器 # timer.start() # tem = result[0]['temperature'] # 23.18 # return result[0] # 应当返回这个字典,再按需取值# 查询结果转为字典def class_to_dict(obj): is_list = obj.__class__ == [].__class__ is_set = obj.__class__ == set().__class__ if is_list or is_set: obj_arr = [] for o in obj: dict = {} a = o.__dict__ if "_sa_instance_state" in a: del a['_sa_instance_state'] dict.update(a) obj_arr.append(dict) return obj_arr else: dict = {} a = obj.__dict__ if "_sa_instance_state" in a: del a['_sa_instance_state'] dict.update(a) return dict# 将json时间格式化class DateEncoder(json.JSONEncoder): def default(self, obj): if isinstance(obj, datetime.datetime): return obj.strftime("%Y-%m-%d %H:%M:%S") else: return json.JSONEncoder.default(self, obj)if __name__ == '__main__': # print(query()) # 创建一个定时器,在程序运行在之后我开启一个insert函数 t1 = threading.Timer(5, function=insert) # 第一个参数是时间,例:过5s之后我执行后面的一个函数,开启一个线程 t1.start() # print(query()) # t2 = threading.Timer(5, function=query) # 第一个参数是时间,例:过5s之后我执行后面的一个函数,开启一个线程 # t2.start() # # 设置一个多线程 # while True: # time.sleep(10) # 延时10s # print('主线程')
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
- 89
- 90
- 91
- 92
- 93
- 94
- 95
- 96
- 97
- 98
- 99
- 100
- 101
- 102
- 103
- 104
- 105
- 106
- 107
- 108
- 109
- 110
- 111
- 112
- 113
- 114
- 115
- 116
- 117
- 118
- 119
- 120
- 121
- 122
- 123
- 124
- 125
- 126
- 127
- 128
- 129
- 130
- 131
- 132
- 133
前端test_dbchart.html:
<!DOCTYPE html><html lang="en"><head> <meta charset="UTF-8"> <title></title> <script type="text/javascript" src="//cdn.bootcss.com/jquery/3.1.1/jquery.min.js"></script> <script type="text/javascript" src="//cdn.bootcss.com/socket.io/1.5.1/socket.io.min.js"></script> <!-- ECharts 引入 --> <script src="https://cdn.staticfile.org/echarts/4.3.0/echarts.min.js"></script></head><body><div id="main" style="height:500px;border:1px solid #ccc;padding:10px;"></div> <script type="text/javascript"> var myChart = echarts.init(document.getElementById('main')); myChart.setOption({ title: { text: '系统监控走势图' }, tooltip: {}, legend: { data:['temperature'] }, xAxis: { data: [] }, yAxis: {}, series: [{ name: 'temperature', type: 'line', data: [] }] }); var record_t = ["","","","","","","","","",""], temperature = [0,0,0,0,0,0,0,0,0,0] //准备好统一的 callback 函数 var update_mychart = function (res) { //res是json格式的response对象 // 隐藏加载动画 myChart.hideLoading(); // 准备数据 record_t.push(res.data[0]); temperature.push(parseFloat(res.data[1])); console.log(temperature) if (record_t.length >= 10){ record_t.shift(); temperature.shift(); } // 填入数据 myChart.setOption({ xAxis: { data: record_t }, series: [{ name: 'temperature', // 根据名字对应到相应的系列 data: temperature }] }); }; // 首次显示加载动画 myChart.showLoading(); // 建立socket连接,等待服务器“推送”数据,用回调函数更新图表 $(document).ready(function() { namespace = '/test'; var socket = io.connect(location.protocol + '//' + document.domain + ':' + location.port + namespace); socket.on('server_response', function(res) { update_mychart(res); }); }); </script></body></html>
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
- 89
后端运行不分顺序,实现如视频:
结语
websocket相对于ajax在实时监控等场景可以较好的应用,相比较于我之前写的ajax数据传输时延大大降低,后续两个工作:
1、物联网接收数据至服务器(数据库)后,自动向客户端发送数据形成成监控和统计图表。
2、进一步写前后端分离的vue+flask+websocket实现。
如果对您有帮助的话,点个赞呗,欢迎各位老爷打赏!