簡介
后端向前端推送信息，通知任務(wù)完成
| 輪詢 | SSE | WebSocket | |
|---|---|---|---|
| 請求方式 | HTTP | HTTP | TCP長連接 |
| 觸發(fā)方式 | 輪詢 | 事件 | 事件 |
| 優(yōu)點 | 實現(xiàn)簡單易兼容 | 實現(xiàn)簡單開發(fā)成本低 | 全雙工通信，開銷小，安全，可擴展 |
| 缺點 | 消耗較大 | 不兼容IE | 傳輸數(shù)據(jù)需二次解析，開發(fā)成本大 |
| 適用場景 | 服務(wù)端向客戶端單向推送 | 網(wǎng)絡(luò)游戲、銀行交互、支付 |
文章來源地址http://www.zghlxwxcb.cn/news/detail-418147.html
安裝
pip install flask
文章來源：http://www.zghlxwxcb.cn/news/detail-418147.html
輪詢
main.py
import time
import threading
from flask_cors import CORS
from flask import Flask, redirect
app = Flask(__name__)
cors = CORS(app)
job = {} # 任務(wù)狀態(tài)
def do_job(id):
global job
job[id] = 'doing'
time.sleep(5)
job[id] = 'done'
@app.route('/job/<id>', methods=['POST'])
def create(id):
"""創(chuàng)建任務(wù)"""
threading.Thread(target=do_job, args=(id,)).start()
response = redirect(f'/job/{id}') # 重定向到查詢該任務(wù)狀態(tài)
return response
@app.route('/job/<id>', methods=['GET'])
def status(id):
"""查詢?nèi)蝿?wù)狀態(tài)"""
return job.get(id, 'not exist')
if __name__ == '__main__':
app.run()
index.html
<!DOCTYPE html>
<html lang="zh">
<head>
<meta charset="UTF-8">
<title>輪詢</title>
<script src="https://cdn.staticfile.org/jquery/3.6.1/jquery.min.js"></script>
</head>
<body>
<button id="create">執(zhí)行任務(wù)</button>
</body>
<script>
$("#create").click(function () {
var id = parseInt(Math.random() * 100000000); // 任務(wù)ID
$.post({
url: "http://127.0.0.1:5000/job/" + id.toString(),
success: function (response) {
$("body").append("<p id='p" + id.toString() + "'>任務(wù)" + id.toString() + "：created</p>");
var interval = setInterval(function () {
$.get({
url: "http://127.0.0.1:5000/job/" + id.toString(),
success: function (response) {
console.log(response);
$("#p" + id.toString()).text("任務(wù)" + id.toString() + "：" + response)
if (response === 'done') {
clearInterval(interval);
}
}
});
}, 1000);
}
});
});
</script>
</html>
效果
SSE
需要異步啟動 + Redis
gunicorn 無法在 Windows 上運行，WSL 對 gevent 支持不友好，建議在純 Linux 系統(tǒng)下使用
安裝 Redis
sudo apt update
sudo apt install redis-server
安裝
pip install flask-sse gunicorn gevent
sse.py
from flask import Flask, render_template
from flask_sse import sse
from flask_cors import CORS
app = Flask(__name__)
app.config['REDIS_URL'] = 'redis://localhost'
app.register_blueprint(sse, url_prefix='/stream')
cors = CORS(app)
@app.route('/')
def index():
return render_template('index.html')
@app.route('/hello')
def publish_hello():
sse.publish({'message': 'Hello!'}, type='greeting')
return 'Message sent!'
templates/index.html
<!DOCTYPE html>
<html lang="zh">
<head>
<meta charset="UTF-8">
<title>SSE</title>
<script src="https://cdn.staticfile.org/jquery/3.6.1/jquery.min.js"></script>
</head>
<body>
<h1>Flask-SSE Quickstart</h1>
<script>
var source = new EventSource("stream");
source.addEventListener("greeting", function (event) {
var data = JSON.parse(event.data);
console.log(data.message)
$("body").append("<p>" + data.message + "</p>");
}, false);
source.addEventListener("error", function (event) {
console.log("Failed to connect to event stream. Is Redis running?");
}, false);
</script>
</body>
</html>
啟動
gunicorn sse:app --worker-class gevent --bind 127.0.0.1:8000
nginx 配置
location ^~ /sse/ {
proxy_pass http://127.0.0.1:8000/;
proxy_set_header Connection '';
proxy_http_version 1.1;
chunked_transfer_encoding off;
}
效果
WebSocket
異步啟動，eventlet 性能最好，然后是 gevent
安裝
pip install flask-socketio gunicorn eventlet
或
pip install flask-socketio gunicorn gevent-websocket
main.py
from flask_socketio import SocketIO
from flask import Flask, render_template, request
app = Flask(__name__)
socketio = SocketIO(app, cors_allowed_origins='*')
connected_sids = set() # 存放已連接的客戶端
@app.route('/')
def index():
return render_template('index.html')
@socketio.on('connect')
def on_connect():
connected_sids.add(request.sid)
print(f'{request.sid} 已連接')
@socketio.on('disconnect')
def on_disconnect():
connected_sids.remove(request.sid)
print(f'{request.sid} 已斷開')
@socketio.on('message')
def handle_message(message):
"""收消息"""
data = message['data']
print(f'{request.sid} {data}')
@app.route('/hello', defaults={'sid': None})
@app.route('/hello/<sid>')
def hello(sid):
"""發(fā)消息"""
if sid:
if sid in connected_sids:
socketio.emit('my_response', {'data': f'Hello, {sid}!'}, room=sid)
return f'已發(fā)信息給{sid}'
else:
return f'{sid}不存在'
else:
socketio.emit('my_response', {'data': 'Hello!'})
return '已群發(fā)信息'
if __name__ == '__main__':
socketio.run(app)
templates/index.html
<!DOCTYPE html>
<html lang="zh">
<head>
<meta charset="UTF-8">
<title>WebSocket</title>
<script src="https://cdn.staticfile.org/jquery/3.6.1/jquery.min.js"></script>
<script src="https://cdn.staticfile.org/socket.io/4.5.2/socket.io.min.js"></script>
</head>
<body>
<h1>Flask-SocketIO Quickstart</h1>
<h2 id="sid">客戶端</h2>
<h2>發(fā)消息</h2>
<input id="emit_data" value="Hello World!">
<button id="emit">發(fā)消息</button>
<h2>收消息</h2>
<div id="log"></div>
<script>
var socket = io();
// var socket = io("ws://127.0.0.1:5000");
socket.on("connect", function () {
$("#sid").text("客戶端：" + socket.id)
});
$("#emit").click(function () {
socket.emit("message", {data: $("#emit_data").val()});
}); // 點擊按鈕發(fā)消息
socket.on("my_response", function (msg) {
$("#log").append("<p>" + msg.data + "</p>"); // 收消息
});
</script>
</body>
</html>
效果
更多內(nèi)容查閱官方示例
WebSocekt 是 HTML5 規(guī)范的一部分，是一種應(yīng)用層協(xié)議，借鑒了 socket 思想，為客戶端和服務(wù)端之間提供了雙向通信功能，包含一套標準的 API。
Socket.IO 是一個 JavaScript 庫，不僅支持 WebSocket，還支持許多種輪詢機制，當 Socket.IO 檢測到當前環(huán)境不支持 WebSocket 時，能自動選擇最佳方式實現(xiàn)網(wǎng)絡(luò)實時通信。
后端 Flask-Sockets 對應(yīng)前端使用原生 WebSocekt
后端 Flask-SocketIO 對應(yīng)前端使用 Socket.IO（推薦這種）
事件
- error：
- reconnect：
- reconnect_attempt：
- reconnect_error：
- reconnect_failed：
- ping：
- connect：
- disconnect：
- connect_error：
參考文獻
- Flask-SSE Documentation
- Flask-SocketIO Documentation
- 長連接/websocket/SSE等主流服務(wù)器推送技術(shù)比較
- Server-Sent Events 與 WebSocket 的比較
- 七牛云CDN
- js停止setInterval的方法與setInterval循環(huán)執(zhí)行的注意事項
- Python flaks-sse 庫的簡單測試
- Ubuntu用命令行打開網(wǎng)頁的三種方法
- 如何使用W3M從Linux終端瀏覽
- 3種 Linux 命令行中使用的 Web 瀏覽器
- EventSource / Server-Sent Events through Nginx
- Server-Sent Events connection timeout on Node.js via Nginx
- Flask教程(十九)SocketIO
- flask-socketio筆記
- websocket在線測試
- WebSocket 教程 - 阮一峰的網(wǎng)絡(luò)日志
- 手摸手教你使用WebSocket
- Web端即時通訊技術(shù)盤點：短輪詢、Comet、Websocket、SSE
- Client API | Socket.IO
- WebSocket 與 Socket.IO
- WebSocket - Web API 接口參考 | MDN
- 在flask上使用websocket
- WebSocket詳解（一）：初步認識WebSocket技術(shù)
- Flask：使用SocketIO實現(xiàn)WebSocket與前端Vue進行實時推送
- Flask使用flask_socketio將信息時時推送前臺
- 使用 Flask-SocketIO 實現(xiàn)私聊：通過flask-socketio中的sid給指定的客戶端發(fā)送消息，對方接收不到
- Flask route parameters default values
到了這里，關(guān)于Python Flask 后端向前端推送信息——輪詢、SSE、WebSocket的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
