Flask-SocketIO使用详解，附实例代码
介绍：flask-socketio模块实际上是封装了flask对websocket的支持，websocket在连接建立阶段是通过HTTP的握手方式进行的，这可以看做是为了兼容浏览器或者使用一些现成的功能来实现，这样一种捷径。当连接建立之后，客户端和服务端之间就不再进行HTTP通信了，所有信息交互都由websocket接管。Flask-SocketIO使Flask应用程序可以访问客户端和服务器之间的低延迟双向通信，使客户端建立与服务器的永久连接。
具体到websocket的实现原理，可以看看这篇文章：websocket原理及使用场景介绍
适用的场景：后台产生新的数据，需要在前台页面马上展示出来，例如数据监控、统计图实时变化更新等。
当然，我们可以使用ajax来完成，通过ajax使得前台定时去后台索要数据，但如果消息频繁，ajax需要不断的建立和释放连接，效果明显不如后端直接推送数据到前台更加合适。
Flask-SocketIO的使用
首先安装依赖：
pip install flask-socketio一个简单的示例（对Flask代码加了一层包装）：
from flask import Flask, render_template
from flask_socketio import SocketIO
app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret!'
socketio = SocketIO(app)
if __name__ == '__main__':
socketio.run(app)SocketIO发送消息
SocketIO可以使用send（）和emit（）函数向连接的客户端发送消息，两个函数有些区别，send（）用于发送未命名事件消息，而emit（）用于发送已命名事件消息。
实例代码：
@socketio.on('message')
def handle_message(message):
send(message, namespace='/chat')
@socketio.on('my event')
def handle_my_custom_event(json):
emit('my response', json, namespace='/chat')namespace表示传入消息的命名空间，前台可以对应这个命名空间选择接收消息，如：
$(document).ready(function() {
namespace = '/test';
var socket = io.connect(location.protocol
+ '//' + document.domain + ':'
+ location.port + namespace);
socket.on('server_response', function(res) {
//res表示接收的数据，这里做数据的处理
});
});一个简单使用SocketIO的完整实例
功能：后台五秒随机产生十个数字，在前台模版中动态刷新显示
后台代码：
#encoding:utf-8
#!/usr/bin/env python
from flask import Flask, render_template
from flask_socketio import SocketIO
import random
async_mode = None
app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret!'
socketio = SocketIO(app)
@app.route('/')
def index():
return render_template('test.html')
@socketio.on('connect', namespace='/test_conn')
def test_connect():
while True:
socketio.sleep(5)
t = random_int_list(1, 100, 10)
socketio.emit('server_response',
{'data': t},
namespace='/test_conn')
def random_int_list(start, stop, length):
start, stop = (int(start), int(stop)) if start <= stop else (int(stop), int(start))
length = int(abs(length)) if length else 0
random_list = []
for i in range(length):
random_list.append(random.randint(start, stop))
return random_list
if __name__ == '__main__':
socketio.run(app, debug=True)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) {
var t = res.data;
$("#t").text(t);
});
});
</script>
</body>
</html>最后，我们使用SocketIO结合Echarts实现一个简单的实时监控图
效果如图所示：
后台代码：
#encoding:utf-8
#!/usr/bin/env python
import psutil
import time
from threading import Lock
from flask import Flask, render_template
from flask_socketio import SocketIO
async_mode = None
app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret!'
socketio = SocketIO(app, async_mode=async_mode)
thread = None
thread_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('index.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__':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 3 引入 -->
<script src="http://echarts.baidu.com/dist/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>目前GoEasy提供完整的websocket前后端解决方案，简单的几行代码集成，即可快速搭建您的专属websocket服务。
同时GoEasy还支持更多的前端技术框架比如小程序、react、vue、uniapp等。同时还是支持php、java、python等服务端语言。 有websocket使用需求的开发者可以来注册GoEasy账号进行测试使用。【立即注册】
————————————————
版权声明：本文为CSDN博主「阿木侠」的原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/weixin_36380516/article/details/80418354