Flask SocketIO 實現動態繪圖
2023-11-27 21:01:11
Flask-SocketIO 是基於 Flask 的一個擴充套件，用於簡化在 Flask 應用中整合 WebSocket 功能。WebSocket 是一種在使用者端和伺服器之間實現實時雙向通訊的協定，常用於實現實時性要求較高的應用，如聊天應用、實時通知等，使得開發者可以更輕鬆地構建實時性要求較高的應用。通過定義事件處理常式，可以實現雙向實時通訊，為應用提供更加豐富和實時的使用者體驗。
前端引數拼接
Flask 提供了針對WebSocket的支援外掛flask_socketio直接通過pip命令安裝即可匯入使用，同時前端也需要引入SocketIO.js庫檔案。
如下程式碼通過ECharts圖表庫和WebSocket技術實現了一個實時監控主機CPU負載的動態折線圖。通過WebSocket連線到Flask應用中的Socket.IO名稱空間，前端通過實時接收後端傳來的CPU負載資料，動態更新折線圖，展示1分鐘、5分鐘和15分鐘的CPU負載趨勢。同時，通過控制檯列印實時資料，實現了方便的偵錯和監控功能。
<!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>
<!-- 執行繪圖函數-->
<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 利用表動態監控',
},
// 調節大小
grid: {
left: '3%',
right: '4%',
bottom: '3%',
containLabel: true
},
// tooltip 滑鼠放上去之後會自動出現座標
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>
<!-- 負責對引數的解析,填充資料 -->
<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("時間陣列: " + time);
console.log("1分鐘: " + cpu_load1);
console.log("5分鐘: " + cpu_load5);
console.log("15分鐘: " + cpu_load15);
// 呼叫繪圖函數
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"}); // 初始化完成後,傳送一條訊息.
socket.on('response', function(recv) {
console.log("時間: " + recv.datetime);
console.log("1分鐘: " + recv.load1);
console.log("5分鐘: " + recv.load5);
console.log("15分鐘: " + recv.load15);
// 呼叫函數完成資料填充
update_function(recv);
});
});
</script>
</body>
</html>
後臺程式碼使用Flask和Flask-SocketIO搭建了一個實時監控主機CPU負載的WebSocket應用，並將資料通過socketio.emit函數將資料推播給前端展示。
關鍵點概括如下：
Flask和SocketIO整合：
- 使用Flask框架建立了一個Web應用，並通過Flask-SocketIO整合了WebSocket功能，實現了實時雙向通訊。
訊息接收與實時推播：
- 定義了
socket事件處理常式，用於接收前端通過WebSocket傳送的訊息。在無限迴圈中，通過socketio.sleep方法設定每2秒推播一次實時的CPU負載資料給前端。
前端連線和斷開事件：
- 定義了
connect和disconnect事件處理常式，分別在WebSocket連線建立和斷開時觸發。在控制檯列印相應資訊，用於監控連線狀態。
實時資料推播：
- 使用
socketio.emit方法實時將CPU負載資料推播給前端，以更新折線圖。推播的資料包括當前時間、1分鐘負載、5分鐘負載和15分鐘負載。
前端頁面渲染：
- 通過Flask的
render_template方法渲染了一個HTML頁面，用於展示實時更新的CPU負載折線圖。
偵錯資訊輸出：
- 在每個事件處理常式中使用
print語句輸出偵錯資訊，方便監測WebSocket連線和訊息的傳遞過程。
總體來說，該應用實現了一個簡單而實用的實時監控系統，通過WebSocket技術實時推播主機CPU負載資料至前端，為使用者提供了實時視覺化的監控體驗。
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")
# 出現訊息後,率先執行此處
@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連線成功時,自動觸發connect預設方法
@socketio.on("connect",namespace="/Socket")
def connect():
print("連結建立成功..")
# 當websocket連線失敗時,自動觸發disconnect預設方法
@socketio.on("disconnect",namespace="/Socket")
def disconnect():
print("連結建立失敗..")
if __name__ == '__main__':
socketio.run(app,debug=True)
執行後，即可輸出當前系統下CPU的負載情況，如下圖所示；
後端引數拼接
如上所示的程式碼是在前端進行的資料拼接，如果我們想要在後端進行資料的拼接，則需要對程式碼進行一定的改進。
前端編寫以下程式碼，通過WebSocket建立通訊隧道，而後臺則每隔2秒向前臺推播傳遞字典資料。
<!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>
<!-- 執行繪圖函數-->
<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 利用表動態監控',
},
// 調節大小
grid: {
left: '3%',
right: '4%',
bottom: '3%',
containLabel: true
},
// tooltip 滑鼠放上去之後會自動出現座標
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"}); // 初始化完成後,傳送一條訊息.
socket.on('response', function(recv) {
console.log("時間: " + recv.datetime);
console.log("1分鐘: " + recv.load1);
console.log("5分鐘: " + recv.load5);
console.log("15分鐘: " + recv.load15);
// 呼叫繪圖函數
display(recv.datetime,recv.load1,recv.load5,recv.load15);
});
});
</script>
</body>
</html>
後臺程式碼則是收集資料，並將資料通過socketio.emit函數，推播給前端。
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)
# 填充資料表
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")
# 出現訊息後,率先執行此處
@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)
# 實現陣列最大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連線成功時,自動觸發connect預設方法
@socketio.on("connect",namespace="/Socket")
def connect():
print("連結建立成功..")
# 當websocket連線失敗時,自動觸發disconnect預設方法
@socketio.on("disconnect",namespace="/Socket")
def disconnect():
print("連結建立失敗..")
if __name__ == '__main__':
socketio.run(app,debug=True)
執行動態圖如下所示；
