Python flaks-sse 库的简单测试_flask sse

作者：运维做开发 | 2024-07-25 12:51:48

踩

flask sse

介绍

服务器推送事件： Server-Sent Events, SSE

特点

仅从服务器向客户端实现单向实时通信。
实现简单，基于HTTP协议。
浏览器端有断线重连功能。
支持用户自定义消息类型。
用来传送文本，二进制需要编码。

flask-sse 实现

flask-sse文档

前置条件

本地Redis服务器
gunicorn gevent, sse是无限事件流，flask处理HTTP请求一次只能响应一个，要需要配合异步服务器使用。

实现

使用官方例子

服务端 sse.py

from flask import Flask, render_template
from flask_sse import sse

app = Flask(__name__)
app.config["REDIS_URL"] = "redis://localhost"
app.register_blueprint(sse, url_prefix='/stream')

@app.route('/')
def index():
    return render_template("index.html")

@app.route('/hello')
def publish_hello():
    sse.publish({"message": "Hello!"}, type='greeting')
    return "Message sent!"
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

客户端
templates/index.html

<!DOCTYPE html>
<html>
<head>
  <title>Flask-SSE Quickstart</title>
</head>
<body>
  <h1>Flask-SSE Quickstart</h1>
  <script>
    var source = new EventSource("{{ url_for('sse.stream') }}");
    source.addEventListener('greeting', function(event) {
        var data = JSON.parse(event.data);
        console.log("The server says " + data.message);
    }, false);
    source.addEventListener('error', function(event) {
        console.log("Failed to connect to event stream. Is Redis running?");
    }, false);
  </script>
</body>
</html>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

gunicorn sse:app --worker-class gevent --bind 127.0.0.1:8000

效果展示

在这里插入图片描述

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/运维做开发/article/detail/880293