轮询
特点：每隔一段时间不短向后端发送请求
缺点：消耗大，延迟
from flask import Flask,render_template,request,jsonify app = Flask(__name__) USERS = { 1:{"name":"Alex","count":250}, 2:{"name":"yuanhao","count":250}, 3:{"name":"bossjin","count":250}, } @app.route("/") def index(): return render_template("index.html",user=USERS) @app.route("/vote",methods=["POST"]) def vote(): uid = request.json.get("uid") USERS[uid]["count"] += 1 return "投票成功" @app.route("/get_vote") def get_vote(): # 返回所有的users数据 return jsonify(USERS) if __name__ == '__main__': app.run()
<!DOCTYPE html> <html lang="zh-CN"> <head> <meta charset="UTF-8"> <title>Title</title> </head> <body> <h1>投票系统</h1> <ul> {% for (key,value) in users.items() %} <li id="{{key}}" onclick="vote({{key}})">{{value.name}}:({{value.count}})</li> {% endfor %} </ul> <script> function vote(uid) { //向后端发送投票请求 带user_id axios.request({ url:"/vote", method:"POST", data:{ uid:uid } }).then(function (res) { console.log(res) }) } function get_vote() { axios.request({ url:"/get_vote", method:"GET" }).then(function (res) { console.log(res); //向后端发获取实时数据的请求 //拿到数据之后更新页面 users = res.data; for(let uid in users){ //根据key获取li标签,改变innnerText let liEle = document.getElementById(uid) liEle.innerText = `${users[uid]["name"]}:(${users[uid]["count"]})` } }) } window.onload = function () { setInterval(get_vote(),2) } </script> </body> </html>
长轮询
特点：满足实时更新
缺点：消耗也大,但是请求次数比轮询小
实现：
利用queue对象实现请求阻塞
每个请求进来都要生成一个q对象
如果有人投票给所有的q对象put数据
拿数据请求从自己的q对象get数据
from flask import Flask,request,render_template,session,jsonify import queue import uuid app = Flask(__name__) app.secret_key = "xxx" USERS = { 1:{"name":"alex","count":25}, 2:{"name":"yuanhao","count":25}, 3:{"name":"bossjin","count":25}, } Q_DICT = {} @app.route("/") def index(): user_uuid = str(uuid.uuid4()) session["user_uuid"] = user_uuid Q_DICT[user_uuid] = queue.Queue() return render_template("index3.html",users=USERS) @app.route("/vote",methods=["POST"]) def vote(): # 投票 uid = request.json.get("uid") USERS[uid]["count"] += 1 for q in Q_DICT.values(): q.put(USERS) return "投票成功" @app.route("/get_vote") def get_vote(): user_uuid = session.get("user_uuid","") q = Q_DICT[user_uuid] try: ret = q.get(timeout=30) except queue.Empty: ret = "" return jsonify(ret) if __name__ == '__main__': app.run()
<!DOCTYPE html> <html lang="zh-CN"> <head> <meta charset="UTF-8"> <title>Title</title> <script src="https://cdn.bootcss.com/axios/0.19.0-beta.1/axios.js"></script> </head> <body> <h1>投票系统</h1> <ul> {% for(key,val) in users.items() %} <li id="{{key}}" onclick=vote({{key}})>{{val.name}}:({{val.count}})</li> {% endfor %} </ul> <script> function vote(uid) { axios.request({ url:"/vote", method:"POST", data:{ uid:uid, } }).then(function (res) { console.log(res) }) } function get_vote() { axios.request({ url:"get_vote", method:"GET", }).then(function (res) { if (res.data != "") { let users = res.data; for(let uid in users){ liELe = document.getElementById(uid); liELe.innerText = `${users[uid]["name"]}:(${users[uid]["count"]})` } } get_vote() }) } window.onload = function () { get_vote() } </script> </body> </html>
websocket
HTTP协议：短链接，无状态，基于TCP/UDP协议进行传输数据
TCP/UDP：传输协议
socket：套接字,API接口
websocket：H5出的新协议，解决轮询的性能问题
特点：
1.握手，基于HTTP进行握手
2.发送数据加密
3.保持连接不断开
下载安装
pip install gevent-websocket # Flask没有websocket
建立一个支持websocket协议连接
from geventwebsocket.handler import WebSocketHandler from gevent.pywsgi import WSGIServer # 拿到websocket对象 ws = request.environ.get("wsgi.websocket") # 发送数据 ws.send() # json # 接收数据 ws.receive() if __name__ == "__main___" http_server = WSGIServer(('0.0.0.0', 5000), app, handler_class=WebSocketHandler) http_server.serve_forever() # 即支持HTTP 也支持websocket
前端发起websocket连接
let ws = new WebSocket("ws://0.0.0.0:5000") # 前端发送数据 ws.send("xxx") # 接收数据 ws.onmessage = function(event){ data = event.data # 数据类型的转换 }