大家好,又见面了,我是你们的朋友全栈君。如果您正在找激活码,请点击查看最新教程,关注关注公众号 “全栈程序员社区” 获取激活教程,可能之前旧版本教程已经失效.最新Idea2022.1教程亲测有效,一键激活。
Jetbrains全系列IDE稳定放心使用
引入
Web端即时通讯技术:即时通讯技术简单的说就是实现这样一种功能:服务器端可以即时地将数据的更新或变化反应到客户端,例如消息即时推送等功能都是通过这种技术实现的。但是在Web中,由于浏览器的限制,实现即时通讯需要借助一些方法。这种限制出现的主要原因是,一般的Web通信都是浏览器先发送请求到服务器,服务器再进行响应完成数据的现实更新。
实现Web端即时通讯的方法:实现即时通讯主要有四种方式,它们分别是轮询、长轮询(comet)、长连接(SSE)、WebSocket。它们大体可以分为两类,一种是在HTTP基础上实现的,包括短轮询、comet和SSE;另一种不是在HTTP基础上实现是,即WebSocket。下面分别介绍一下这四种轮询方式,以及它们各自的优缺点。
当我们要实现一个实时投票系统,或者是实时通讯系统,我们的页面数据总需要更新
我们不能让用户一直去刷新页面。所以就有了轮询,长轮询,以及websock的出现
轮询
既然我想要实时获取后端的数据,那我就每隔2秒给后端发一次请求
这种我们就叫轮询~那它会有一些缺点就是存在延时~就算每秒发一次~也会存在一定的延迟
下面我们看下轮询的代码:
from flask import Flask, render_template, request, jsonify
app = Flask(__name__)
USERS = {
1: {“name”: “悟空”, “count”: 0},
2: {“name”: “悟能”, “count”: 0},
3: {“name”: “悟净”, “count”: 0},
}
@app.route(“/”)
def index():
return render_template(“index.html”, users=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():
return jsonify(USERS)
if __name__ == ‘__main__’:
app.run()
app.py
Title
选丑大赛
{% for key, value in users.items()%}
{
{value.name}} ({
{value.count}})
{% endfor %}
function vote(uid) {
axios.request({
url: “/vote”,
method: “POST”,
data: {
“uid”: uid
}
}).then(function (res) {
console.log(res.data)
})
}
function get_vote() {
axios.request({
url: “/get_vote”,
method: “GET”
}).then(function (res) {
console.log(res)
for(let key in res.data){
let liEle = document.getElementById(key);
let username = res.data[key][“name”]
let count = res.data[key][“count”]
liEle.innerText = `${username} (${count})`
}
})
}
window.onload = function () {
setInterval(get_vote, 2000)
}
index.html
长轮询
轮询缺点就是延迟,那么如果前端发送过来请求,如果没有数据的更新
后端的请求就阻塞了,直到有数据返回或者超时再返回,这样延迟就可以得到很好的解决
python中有个queue对象,当我们从这个队列里拿不到值的时候,可以阻塞住请求的
import queue
from flask import Flask
app = Flask(__name__)
q = queue.Queue()
@app.route(“/get”)
def index():
try:
val = q.get(timeout=20)
except queue.Empty:
val = “超时”
return val
@app.route(“/vote”)
def vote():
q.put(“10”)
return “投票成功”
if __name__ == ‘__main__’:
app.run()
queueDemo.py
如果我为每个请求都建立一个q对象,然后阻塞住他们的请求,有数据更新的时候,给他们的q对象返回值就可以了。
from flask import Flask, render_template, request, jsonify, session
import queue
import uuid
app = Flask(__name__)
app.secret_key = “lajdgia”
USERS = {
1: {“name”: “悟空”, “count”: 0},
2: {“name”: “悟能”, “count”: 0},
3: {“name”: “悟净”, “count”: 0},
}
# 为每个用户建立一个q对象
# 以用户的uuid为key 值为q对象
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(“index2.html”, users=USERS)
@app.route(“/vote”, methods=[“POST”])
def vote():
# 投票 循环q对象的dict 给每个q对象返回值
uid = request.json.get(“uid”)
USERS[uid][“count”] += 1
for q in Q_DICT.values():
q.put(USERS)
return “投票成功”
@app.route(“/get_vote”, methods=[“POST”, “GET”])
def get_vote():
# 获取投票结果 去自己的q对象里取值 没有夯住 知道有或者超时返回
user_uuid = session.get(“user_uuid”)
q = Q_DICT[user_uuid]
try:
users = q.get(timeout=30)
except queue.Empty:
users = “”
return jsonify(users)
if __name__ == ‘__main__’:
app.run()
app2.py
Title
选丑大赛
{% for key, value in users.items()%}
{
{value.name}} ({
{value.count}})
{% endfor %}
function vote(uid) {
axios.request({
url: “/vote”,
method: “POST”,
data: {
“uid”: uid
}
}).then(function (res) {
console.log(res.data)
})
}
function get_votes() {
axios.request({
url: “/get_vote”,
method: “POST”
}).then(function (res) {
console.log(res);
if(res.data != “”){
for(let key in res.data){
let liEle = document.getElementById(key);
let username = res.data[key][“name”]
let count = res.data[key][“count”]
liEle.innerText = `${username} (${count})`
}
}
get_votes()
})
}
window.onload = function () {
get_votes()
}
index2.html
websocket
websocket是一个协议,协议规定
连接的时候需要握手,发送的数据需要加密~~连接之后不断开
Flask不带websocket,我们需要下载
下载:pip install gevent-websocket
from flask import Flask, request, render_template
from geventwebsocket.handler import WebSocketHandler
from gevent.pywsgi import WSGIServer
import json
app = Flask(__name__)
USERS = {
1: {“name”: “悟空”, “count”: 0},
2: {“name”: “悟能”, “count”: 0},
3: {“name”: “悟净”, “count”: 0},
}
@app.route(“/”)
def index():
return render_template(“index3.html”, users=USERS)
WEBSOCKET_LIST = []
@app.route(“/vote”)
def vote():
ws = request.environ.get(“wsgi.websocket”)
if not ws:
return “HTTP请求”
WEBSOCKET_LIST.append(ws)
while True:
uid = ws.receive()
if not uid:
WEBSOCKET_LIST.remove(ws)
ws.close()
break
uid = int(uid)
USERS[uid][“count”] += 1
name = USERS[uid][“name”]
new_count = USERS[uid][“count”]
for client in WEBSOCKET_LIST:
client.send(json.dumps({“uid”: uid, “name”: name, “count”: new_count}))
if __name__ == ‘__main__’:
http_server = WSGIServer((‘127.0.0.1’, 5000), app, handler_class=WebSocketHandler)
http_server.serve_forever()
app3.py
Title
选丑大赛
{% for (key, value) in users.items() %}
{
{value.name}} ({
{value.count}})
{% endfor%}
let ws = new WebSocket(‘ws://127.0.0.1:5000/vote’)
function vote(uid) {
ws.send(uid)
}
ws.onmessage = function (event) {
let data = JSON.parse(event.data);
let liEle = document.getElementById(data.uid);
liEle.innerText = `${data.name} (${data.count})`
}
index3.html
作 者:Warm and new
发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/182049.html原文链接:https://javaforall.cn
【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛
【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...