前言
flask-socketio 为flask应用提供了一个客户端与服务器之间低延迟的双向通讯
官网地址：https://flask-socketio.readthedocs.io/en/latest/intro.html
环境准备
先安装flask-socketio
pip install flask-socketio
说明Flask-SocketIO 与 js版本客户端不匹配，二者不能正常通信。兼容版本说明：https://flask-socketio.readthedocs.io/en/latest/intro.html
我安装的5.x的版本
那么前端socket.io.js应该选4.x版本
快速开始
以下是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)
前端简单代码
<script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.0.1/socket.io.js" integrity="sha512-q/dWJ3kcmjBLU4Qc47E4A9kTB4m3wuTY7vkFJDTZKjTs8jhyGQnaUrxa0Ytd0ssMZhbNua9hE+E7Qv1j+DyZwA==" crossorigin="anonymous"></script>
<script type="text/javascript" charset="utf-8">
var socket = io();
socket.on('connect', function() {
socket.emit('my event', {data: 'I\'m connected!'});
});
</script>
前后端交互实例
前端代码
<!DOCTYPE html>
<html>
<head>
<title>web</title>
<meta charset="UTF-8">
<script type="text/javascript" src="/static/jquery-3.5.1.min.js"></script>
<script type="text/javascript" src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.0.1/socket.io.js"></script>
</head>
<body>
<script type="text/javascript" charset="UTF-8">
$(document).ready(function() {
const namespace = '/ws'
// 连上服务器
const socket = io.connect("http://" + document.domain + ":" + location.port + namespace);
// 发送message 消息
socket.emit("message",{"data":"hello world!"});
// 收到数据后,执行输出
socket.on('response', function(recv) {
console.log('hello： ' + recv.data)
});
});
</script>
</body>
</html>
flask 后端代码
from flask import Flask, render_template,request
from flask_socketio import SocketIO
app = Flask(import_name=__name__,
static_url_path='/static', # 配置静态文件的访问url前缀
static_folder='static', # 配置静态文件的文件夹
template_folder='templates') # 配置模板文件的文件夹
app.config['SECRET_KEY'] = "abcdef"
socketio = SocketIO(app)
@app.route("/")
def index():
return render_template("web.html")
# 出现消息后,率先执行此处
@socketio.on("message", namespace="/ws")
def socket(message):
print(f"接收到消息: {message['data']}")
for i in range(1, 10):
socketio.sleep(1)
socketio.emit("response", # 绑定通信
{"data": i}, # 返回socket数据
namespace="/ws")
# 当websocket连接成功时,自动触发connect默认方法
@socketio.on("connect", namespace="/ws")
def connect():
print("链接建立成功..")
# 当websocket连接失败时,自动触发disconnect默认方法
@socketio.on("disconnect", namespace="/ws")
def disconnect():
print("链接建立失败..")
if __name__ == '__main__':
socketio.run(
app,
debug=True,
host="0.0.0.0",
port=8000,
allow_unsafe_werkzeug=True
)
在console看到输出1-9 ，说明前后端交互成功
更多使用方法参考官网教程：https://flask-socketio.readthedocs.io/en/latest/getting_started.html
标签：__,websocket,socket,socketio,Flask,app,99,flask From： https://www.cnblogs.com/yoyoketang/p/18022139