Flask-SocketIO 是基于 Flask 的一個擴展，用于簡化在 Flask 應用中集成 WebSocket 功能。WebSocket 是一種在客戶端和服務器之間實現(xiàn)實時雙向通信的協(xié)議，常用于實現(xiàn)實時性要求較高的應用，如聊天應用、實時通知等，使得開發(fā)者可以更輕松地構建實時性要求較高的應用。通過定義事件處理函數(shù)，可以實現(xiàn)雙向實時通信，為應用提供更加豐富和實時的用戶體驗。
前端參數(shù)拼接
Flask 提供了針對WebSocket的支持插件flask_socketio直接通過pip命令安裝即可導入使用，同時前端也需要引入SocketIO.js庫文件。
如下代碼通過ECharts圖表庫和WebSocket技術實現(xiàn)了一個實時監(jiān)控主機CPU負載的動態(tài)折線圖。通過WebSocket連接到Flask應用中的Socket.IO命名空間，前端通過實時接收后端傳來的CPU負載數(shù)據(jù)，動態(tài)更新折線圖，展示1分鐘、5分鐘和15分鐘的CPU負載趨勢。同時，通過控制臺打印實時數(shù)據(jù)，實現(xiàn)了方便的調(diào)試和監(jiān)控功能。
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<script type="text/javascript" src="https://www.lyshark.com/javascript/jquery/3.5.1/jquery.min.js"></script>
<script type="text/javascript" src="https://www.lyshark.com/javascript/socket.io/socket.io.min.js"></script>
<script type="text/javascript" src="https://www.lyshark.com/javascript/echarts/5.3.0/echarts.min.js"></script>
</head>
<body>
<div id="Linechart" style="height:500px;width:1200px;border:1px solid #673ab7;padding:10px;"></div>
<!-- 執(zhí)行繪圖函數(shù)-->
<script type="text/javascript" charset="UTF-8">
var display = function(time,x,y,z)
{
var echo = echarts.init(document.getElementById("Linechart"));
var option = {
title: {
left: 'left',
text: 'CPU 利用表動態(tài)監(jiān)控',
},
// 調(diào)節(jié)大小
grid: {
left: '3%',
right: '4%',
bottom: '3%',
containLabel: true
},
// tooltip 鼠標放上去之后會自動出現(xiàn)坐標
tooltip: {
trigger: 'axis',
axisPointer: {
type: 'cross',
label: {
backgroundColor: '#6a7985'
}
}
},
legend: {
data: ['1分鐘負載', '5分鐘負載', '15分鐘負載']
},
xAxis: {
type: 'category',
// data: ['Mon', 'Tue', 'Wed', 'Thu', 'Fri', 'Sat', 'Sun']
data: time
},
yAxis: {
type: 'value'
},
series:
[
{
name: "1分鐘負載",
stack: "總量",
//data: [10, 25, 99, 87, 54, 66, 2],
data: x,
type: 'line',
areaStyle: {}
},
{
name: "5分鐘負載",
stack: "總量",
//data: [89, 57, 85, 44, 25, 4, 54],
data: y,
type: 'line',
areaStyle: {}
},
{
name: "15分鐘負載",
stack: "總量",
//data: [1, 43, 2, 12, 5, 4, 7],
data: z,
type: 'line',
areaStyle: {}
}
]
};
echo.setOption(option,true);
}
</script>
<!-- 負責對參數(shù)的解析,填充數(shù)據(jù) -->
<script type="text/javascript" charset="UTF-8">
var time =["","","","","","","","","","","","","","","","","","","","","","","","","","","","","","","","","","",""];
var cpu_load1 = [0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0];
var cpu_load5 = [0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0];
var cpu_load15 = [0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0];
var update_function = function(recv)
{
time.push(recv.datetime);
cpu_load1.push(parseFloat(recv.load1));
cpu_load5.push(parseFloat(recv.load5));
cpu_load15.push(parseFloat(recv.load15));
if(time.length >=10)
{
time.shift();
cpu_load1.shift();
cpu_load5.shift();
cpu_load15.shift();
console.log("時間數(shù)組: " + time);
console.log("1分鐘: " + cpu_load1);
console.log("5分鐘: " + cpu_load5);
console.log("15分鐘: " + cpu_load15);
// 調(diào)用繪圖函數(shù)
display(time,cpu_load1,cpu_load5,cpu_load15);
}
};
</script>
<!-- 負責接收目標主機的CPU負載情況 -->
<script type="text/javascript" charset="UTF-8">
$(document).ready(function()
{
namespace = '/Socket';
var socket = io.connect("http://" + document.domain + ":" + location.port + namespace);
socket.emit("message",{"data":"hello lyshark"}); // 初始化完成后,發(fā)送一條消息.
socket.on('response', function(recv) {
console.log("時間: " + recv.datetime);
console.log("1分鐘: " + recv.load1);
console.log("5分鐘: " + recv.load5);
console.log("15分鐘: " + recv.load15);
// 調(diào)用函數(shù)完成數(shù)據(jù)填充
update_function(recv);
});
});
</script>
</body>
</html>
后臺代碼使用Flask和Flask-SocketIO搭建了一個實時監(jiān)控主機CPU負載的WebSocket應用，并將數(shù)據(jù)通過socketio.emit函數(shù)將數(shù)據(jù)推送給前端展示。
關鍵點概括如下：
Flask和SocketIO集成：
- 使用Flask框架創(chuàng)建了一個Web應用，并通過Flask-SocketIO集成了WebSocket功能，實現(xiàn)了實時雙向通信。
消息接收與實時推送：
- 定義了
socket事件處理函數(shù)，用于接收前端通過WebSocket發(fā)送的消息。在無限循環(huán)中，通過socketio.sleep方法設置每2秒推送一次實時的CPU負載數(shù)據(jù)給前端。
前端連接和斷開事件：
- 定義了
connect和disconnect事件處理函數(shù)，分別在WebSocket連接建立和斷開時觸發(fā)。在控制臺打印相應信息，用于監(jiān)控連接狀態(tài)。
實時數(shù)據(jù)推送：
- 使用
socketio.emit方法實時將CPU負載數(shù)據(jù)推送給前端，以更新折線圖。推送的數(shù)據(jù)包括當前時間、1分鐘負載、5分鐘負載和15分鐘負載。
前端頁面渲染：
- 通過Flask的
render_template方法渲染了一個HTML頁面，用于展示實時更新的CPU負載折線圖。
調(diào)試信息輸出：
- 在每個事件處理函數(shù)中使用
print語句輸出調(diào)試信息，方便監(jiān)測WebSocket連接和消息的傳遞過程。
總體來說，該應用實現(xiàn)了一個簡單而實用的實時監(jiān)控系統(tǒng)，通過WebSocket技術實時推送主機CPU負載數(shù)據(jù)至前端，為用戶提供了實時可視化的監(jiān)控體驗。
from flask import Flask,render_template,request
from flask_socketio import SocketIO
import time,psutil
async_mode = None
app = Flask(__name__)
app.config['SECRET_KEY'] = "lyshark"
socketio = SocketIO(app)
@app.route("/")
def index():
return render_template("index.html")
# 出現(xiàn)消息后,率先執(zhí)行此處
@socketio.on("message",namespace="/Socket")
def socket(message):
print("接收到消息:",message['data'])
while True:
socketio.sleep(2)
data = time.strftime("%M:%S",time.localtime())
cpu = psutil.cpu_percent(interval=None,percpu=True)
socketio.emit("response",
{"datetime": data, "load1": cpu[0], "load5": cpu[1], "load15": cpu[2]},
namespace="/Socket")
# 當websocket連接成功時,自動觸發(fā)connect默認方法
@socketio.on("connect",namespace="/Socket")
def connect():
print("鏈接建立成功..")
# 當websocket連接失敗時,自動觸發(fā)disconnect默認方法
@socketio.on("disconnect",namespace="/Socket")
def disconnect():
print("鏈接建立失敗..")
if __name__ == '__main__':
socketio.run(app,debug=True)
運行后，即可輸出當前系統(tǒng)下CPU的負載情況，如下圖所示；
后端參數(shù)拼接
如上所示的代碼是在前端進行的數(shù)據(jù)拼接，如果我們想要在后端進行數(shù)據(jù)的拼接，則需要對代碼進行一定的改進。
前端編寫以下代碼，通過WebSocket建立通信隧道，而后臺則每隔2秒向前臺推送傳遞字典數(shù)據(jù)。
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<script type="text/javascript" src="https://www.lyshark.com/javascript/jquery/3.5.1/jquery.min.js"></script>
<script type="text/javascript" src="https://www.lyshark.com/javascript/socket.io/socket.io.min.js"></script>
<script type="text/javascript" src="https://www.lyshark.com/javascript/echarts/5.3.0/echarts.min.js"></script>
</head>
<body>
<div id="Linechart" style="height:500px;width:1200px;border:1px solid #673ab7;padding:10px;"></div>
<!-- 執(zhí)行繪圖函數(shù)-->
<script type="text/javascript" charset="UTF-8">
var display = function(time,x,y,z)
{
var echo = echarts.init(document.getElementById("Linechart"));
var option = {
title: {
left: 'left',
text: 'CPU 利用表動態(tài)監(jiān)控',
},
// 調(diào)節(jié)大小
grid: {
left: '3%',
right: '4%',
bottom: '3%',
containLabel: true
},
// tooltip 鼠標放上去之后會自動出現(xiàn)坐標
tooltip: {
trigger: 'axis',
axisPointer: {
type: 'cross',
label: {
backgroundColor: '#6a7985'
}
}
},
legend: {
data: ['1分鐘負載', '5分鐘負載', '15分鐘負載']
},
xAxis: {
type: 'category',
// data: ['Mon', 'Tue', 'Wed', 'Thu', 'Fri', 'Sat', 'Sun']
data: time
},
yAxis: {
type: 'value'
},
series:
[
{
name: "1分鐘負載",
stack: "總量",
//data: [10, 25, 99, 87, 54, 66, 2],
data: x,
type: 'line',
areaStyle: {}
},
{
name: "5分鐘負載",
stack: "總量",
//data: [89, 57, 85, 44, 25, 4, 54],
data: y,
type: 'line',
areaStyle: {}
},
{
name: "15分鐘負載",
stack: "總量",
//data: [1, 43, 2, 12, 5, 4, 7],
data: z,
type: 'line',
areaStyle: {}
}
]
};
echo.setOption(option,true);
}
</script>
<!-- 負責接收目標主機的CPU負載情況 -->
<script type="text/javascript" charset="UTF-8">
$(document).ready(function()
{
namespace = '/Socket';
var socket = io.connect("http://" + document.domain + ":" + location.port + namespace);
socket.emit("message",{"data":"hello lyshark"}); // 初始化完成后,發(fā)送一條消息.
socket.on('response', function(recv) {
console.log("時間: " + recv.datetime);
console.log("1分鐘: " + recv.load1);
console.log("5分鐘: " + recv.load5);
console.log("15分鐘: " + recv.load15);
// 調(diào)用繪圖函數(shù)
display(recv.datetime,recv.load1,recv.load5,recv.load15);
});
});
</script>
</body>
</html>
后臺代碼則是收集數(shù)據(jù)，并將數(shù)據(jù)通過socketio.emit函數(shù)，推送給前端。
from flask import Flask,render_template,request
from flask_socketio import SocketIO
import time,psutil
async_mode = None
app = Flask(__name__)
app.config['SECRET_KEY'] = "lyshark"
socketio = SocketIO(app)
# 填充數(shù)據(jù)表
local_time = ["", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", ""]
cpu_load1 = [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
cpu_load5 = [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
cpu_load15 = [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
# 左移填充
def shift(Array, Size, Push):
if len(Array) <= Size and len(Array) >= 0:
Array.pop(0)
Array.append(Push)
return True
return False
# 右移填充
def unshift(Array, Size, Push):
if len(Array) <= Size and len(Array) >= 0:
Array.pop(Size-1)
Array.insert(0,Push)
@app.route("/")
def index():
return render_template("index.html")
# 出現(xiàn)消息后,率先執(zhí)行此處
@socketio.on("message",namespace="/Socket")
def socket(message):
print("接收到消息:",message['data'])
while True:
socketio.sleep(1)
data = time.strftime("%M:%S",time.localtime())
cpu = psutil.cpu_percent(interval=None,percpu=True)
# 實現(xiàn)數(shù)組最大35,每次左移覆蓋第一個
shift(local_time,35,data)
shift(cpu_load1,35,cpu[0])
shift(cpu_load5, 35, cpu[1])
shift(cpu_load15, 35, cpu[2])
socketio.emit("response",
{"datetime": local_time, "load1": cpu_load1, "load5": cpu_load5, "load15": cpu_load15},
namespace="/Socket")
# 當websocket連接成功時,自動觸發(fā)connect默認方法
@socketio.on("connect",namespace="/Socket")
def connect():
print("鏈接建立成功..")
# 當websocket連接失敗時,自動觸發(fā)disconnect默認方法
@socketio.on("disconnect",namespace="/Socket")
def disconnect():
print("鏈接建立失敗..")
if __name__ == '__main__':
socketio.run(app,debug=True)
運行動態(tài)圖如下所示；文章來源：http://www.zghlxwxcb.cn/news/detail-747745.html
文章來源地址http://www.zghlxwxcb.cn/news/detail-747745.html
到了這里，關于Flask SocketIO 實現(xiàn)動態(tài)繪圖的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關文章，希望大家以后多多支持TOY模板網(wǎng)！
