SocketIO是大名鼎鼎的实时通讯库,可以在服务器和页面之间轻松的实现双向实时通讯, 兼容性好,使用方便.多用来制作聊天/直播室, 等需要实时传输数据的地方. 今天就简单介绍一下在flask中使用SocketIO的方法.
flask 是使用一个较flask-socketio的扩展来实现对SocketIO的实现的.安装方法.
pip3 install flask-socketio
如果你不是用python来做运维的,希望你使用的python 3.x,否则你会无法使用python的很多新特性,毕竟python2.x的技术支持即将停止,而且一些重量级的工具(比如numpy, pandas等)也在新版本中不保证兼容python2.x版本.
例子很简单,是一个实现基本消息发送的页面.
from flask import Flask
from flask import request
from flask import session
from flask_session import Session
import os
from flask import render_template
from flask_socketio import SocketIO
import json
from module.quotations_module import Quotation
port = 8001
secret_key = os.urandom(24) # 生成密钥，为session服务。
app = Flask(__name__)
app.config['SECRET_KEY'] = secret_key # 配置会话密钥
app.config['SESSION_TYPE'] = "redis" # session类型为redis
app.config['SESSION_PERMANENT'] = True # 如果设置为True，则关闭浏览器session就失效
app.config.from_object(__name__)
Session(app)
socket_io = SocketIO(app)
@app.route("/listen", methods=['post', 'get'])
def listen_func():
""""监听发送来的消息,并使用socketio向所有客户端发送消息"""
mes = {"message": "unknown error"}
data = request.args['data'] if request.args.get('data') else request.form.get('data')
if data is not None:
socket_io.emit(data=data, event="mes") // js客户端on绑定的就是这个event的事件
mes['message'] = "success"
else:
pass
return json.dumps(mes)
@socket_io.on("login")
def quotations_func(mes):
"""客户端连接"""
print(mes)
sid = request.sid # io客户端的sid, socketio用此唯一标识客户端.
can = False
host = request.host
host_list = ['127.0.0.1']
"""
根据页面的访问地址决定是否允许连接, 你可以自己实现自己对访问的控制逻辑.
最后决定是允许连接还是使用socket_io.server.disconnect(sid)断开链接.
"""
if host in host_list:
can = True
elif host.startswith("192.168") or host.startswith("local"):
can = True
else:
pass
if can:
socket_io.emit(event="login", data=json.dumps({"message": "connect refuse!"}))
socket_io.server.disconnect(sid)
else:
socket_io.emit(event="login", data=json.dumps({"message": "connect success!"}))
@app.route("/test")
def test_func():
"""测试页面"""
return render_template("test.html")
if __name__ == '__main__':
socket_io.run(app=app, host="0.0.0.0", port=port, debug=True)
测试页面的代码
<!DOCTYPE html>
<html lang="zh-cn">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<style>
ul{
text-align: left;
margin-left: 10px;
list-style-type: none;
}
body{
text-align: center;
}
#show, #bottom{
width: 600px;
background-color: azure;
padding: 15px
}
#show{
height: 400px;
}
#bottom{
height: 30px;
}
input{
width: 520px;
}
</style>
<body>
<div id="show">
<ul>
</ul>
</div>
<div id="bottom">
<input id="text"><button id="submit">发送</button>
</div>
</body>
<script src="/static/js/jquery.js"></script> <!--jquery请自行百度下载-->
<script src="/static/js/socket.io.js"></script> <!--socket.io.js请自行百度下载-->
<script>
$(function(){
var url = "http://" + document.domain + ':' + location.port;
console.log(url);
var io_client = io.connect(url);
io_client.on('connect', function() {
// 连接成功时的事件
io_client.emit('login', {data: 'I\'m connected!'});
});
io_client.on("mes", function(resp){
// 绑定的事件, 对应py文件中的event参数的值
$("#show ul").append(`<li>${resp}</li>`);
});
// 发送按钮事件
$("#submit").click(function(){
var text = $.trim($("#text").val());
if(text == ""){
// nothing...
}
else{
$.post("/listen", {"data": text}, function(resp){
var resp = JSON.parse(resp);
var status = resp['message'];
if(status == "success"){
$("#text").val(""); // 清空输入内容
}
else{
alert(status);
}
});
}
});
// end !!!
});
</script>
</html>
打开2个 http://127.0.0.1:8001/test页面, 输入点字符然后回车看看.
