Python Flask 后端向前端推送信息——轮询、SSE、WebSocket-Toy模板网

这篇具有很好参考价值的文章主要介绍了Python Flask 后端向前端推送信息——轮询、SSE、WebSocket。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

简介

后端向前端推送信息，通知任务完成

	轮询	SSE	WebSocket
请求方式	HTTP	HTTP	TCP长连接
触发方式	轮询	事件	事件
优点	实现简单易兼容	实现简单开发成本低	全双工通信，开销小，安全，可扩展
缺点	消耗较大	不兼容IE	传输数据需二次解析，开发成本大
适用场景		服务端向客户端单向推送	网络游戏、银行交互、支付

文章来源地址https://www.toymoban.com/news/detail-418147.html

安装

pip install flask

轮询

main.py

import time
import threading

from flask_cors import CORS
from flask import Flask, redirect

app = Flask(__name__)
cors = CORS(app)

job = {}  # 任务状态


def do_job(id):
    global job
    job[id] = 'doing'
    time.sleep(5)
    job[id] = 'done'


@app.route('/job/<id>', methods=['POST'])
def create(id):
    """创建任务"""
    threading.Thread(target=do_job, args=(id,)).start()
    response = redirect(f'/job/{id}')  # 重定向到查询该任务状态
    return response


@app.route('/job/<id>', methods=['GET'])
def status(id):
    """查询任务状态"""
    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">执行任务</button>
</body>
<script>
    $("#create").click(function () {
        var id = parseInt(Math.random() * 100000000);  // 任务ID
        $.post({
            url: "http://127.0.0.1:5000/job/" + id.toString(),
            success: function (response) {
                $("body").append("<p id='p" + id.toString() + "'>任务" + 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("任务" + id.toString() + "：" + response)
                            if (response === 'done') {
                                clearInterval(interval);
                            }
                        }
                    });
                }, 1000);
            }
        });
    });
</script>
</html>

效果

Python Flask 后端向前端推送信息——轮询、SSE、WebSocket

SSE

需要异步启动 + Redis

gunicorn 无法在 Windows 上运行，WSL 对 gevent 支持不友好，建议在纯 Linux 系统下使用

安装 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;
}

效果

Python Flask 后端向前端推送信息——轮询、SSE、WebSocket

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):
    """发消息"""
    if sid:
        if sid in connected_sids:
            socketio.emit('my_response', {'data': f'Hello, {sid}!'}, room=sid)
            return f'已发信息给{sid}'
        else:
            return f'{sid}不存在'
    else:
        socketio.emit('my_response', {'data': 'Hello!'})
        return '已群发信息'


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>发消息</h2>
<input id="emit_data" value="Hello World!">
<button id="emit">发消息</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()});
    });  // 点击按钮发消息

    socket.on("my_response", function (msg) {
        $("#log").append("<p>" + msg.data + "</p>");  // 收消息
    });
</script>
</body>
</html>

效果

Python Flask 后端向前端推送信息——轮询、SSE、WebSocket

更多内容查阅官方示例

WebSocekt 是 HTML5 规范的一部分，是一种应用层协议，借鉴了 socket 思想，为客户端和服务端之间提供了双向通信功能，包含一套标准的 API。

Socket.IO 是一个 JavaScript 库，不仅支持 WebSocket，还支持许多种轮询机制，当 Socket.IO 检测到当前环境不支持 WebSocket 时，能自动选择最佳方式实现网络实时通信。

后端 Flask-Sockets 对应前端使用原生 WebSocekt
后端 Flask-SocketIO 对应前端使用 Socket.IO（推荐这种）