Flask-SSE: 实时服务器推送|极客教程

Flask-SSE: 实时服务器推送

一、介绍

在传统的 Web 应用中，用户向服务器发送请求，服务器根据请求返回相应的内容。然而，随着互联网的快速发展，用户对于实时性的需求也越来越高。传统的请求-响应方式无法满足这种需求。在这种情况下，服务器推送（Server Sent Events，简称 SSE）成为了一种重要的技术手段，用于实现服务器主动向客户端推送数据。

Flask-SSE 是一个基于 Python 的轻量级框架，用于在 Flask 应用中实现 SSE 功能。本文将详细介绍 Flask-SSE 的使用方法，帮助读者更好地理解和应用 SSE 技术。

二、安装

在开始使用 Flask-SSE 之前，我们需要先安装它。可以通过 pip 包管理工具进行安装：

pip install flask-sse

三、基本用法

首先，我们需要在 Flask 应用中引入 Flask-SSE：

from flask import Flask
from flask_sse import sse

app = Flask(__name__)
app.register_blueprint(sse, url_prefix='/stream')

在上面的代码中，我们创建了一个 Flask 应用，并注册了 Flask-SSE 扩展。同时，我们将 SSE 的 URL 前缀设置为 “/stream”，这意味着 SSE 将通过这个 URL 提供服务。

接下来，我们可以定义一个视图函数，用于接收 SSE 的连接：

@app.route('/')
def index():
    return "Hello SSE!"

@app.route('/stream')
def stream():
    return sse.stream()

在上面的代码中，我们创建了两个视图函数。”/” 路由返回一个简单的字符串作为欢迎信息，用于测试 SSE 是否正常工作。”/stream” 路由则返回 sse.stream()，这个函数用于建立 SSE 连接。

此时，我们可以启动 Flask 应用，然后使用浏览器访问 “http://localhost:5000/stream”，即可与 SSE 服务器建立连接。

四、发送事件

在建立 SSE 连接之后，我们可以向客户端发送事件。每个事件由一个标识符和数据组成。在 Flask-SSE 中，我们可以使用 send_event() 函数来发送事件。

下面是一个简单的示例，向连接的客户端发送一个名为 “message” 的事件：

from flask_sse import sse

@app.route('/')
def index():
    sse.publish({"message": "Hello, world!"}, type='message_event')
    return "Hello SSE!"

在上面的代码中，我们使用 sse.publish() 函数向客户端发送一个名为 “message_event” 的事件，并附带一条消息作为数据。当客户端收到这个事件时，就可以根据需要处理。

五、接收事件

在客户端连接建立之后，可以通过 JavaScript 来监听并处理来自服务器的事件。Flask-SSE 提供了一种简单易用的 JavaScript API，用于接收 SSE 事件。

下面是一个示例代码，用于接收服务器发送的事件：

var source = new EventSource('/stream');

source.addEventListener('message_event', function(event) {
    var data = JSON.parse(event.data);
    console.log(data.message);
});

在上面的代码中，我们使用 EventSource 对象来建立 SSE 连接，并通过 addEventListener() 方法监听名为 “message_event” 的事件。当服务器发送这个事件时，就会触发监听器，并打印消息内容。

六、多个连接

Flask-SSE 在处理多个连接方面也非常出色。它为每个连接维护一个 SSEStream 对象，可以通过 sse.publish() 方法向所有连接发送事件。

下面是一个示例代码，用于同时向多个客户端发送事件：

from flask_sse import sse

@app.route('/')
def index():
    return "Hello SSE!"

@app.route('/stream')
def stream():
    sse.publish({"message": "Hello, world!"}, type='message_event')
    return sse.stream()

在上面的代码中，我们先返回一个简单的欢迎信息，然后通过 sse.publish() 函数向所有客户端发送一个名为 “message_event” 的事件。请注意，我们只需要在其中一个连接的视图函数中调用 sse.publish() 函数即可。

七、进阶用法

1. 设置重试时间间隔

默认情况下，SSE 连接被关闭后，客户端会在 3 秒后重新连接。如果你想修改这个时间间隔，可以在初始化 SSEStream 对象时通过 retries 参数进行设置。

from flask_sse import sse

@app.route('/')
def index():
    return "Hello SSE!"

@app.route('/stream')
def stream():
    retries = 10
    return sse.stream(retries=retries)

在上面的代码中，我们设置重新连接的时间间隔为 10 秒。

2. 自定义事件名称和数据格式

有时候，我们可能需要自定义事件的名称，以及发送的数据格式。Flask-SSE 提供了相应的方法，以满足不同的需求。

下面是一个示例代码，用于自定义事件名称和数据格式：

from flask_sse import sse

@app.route('/')
def index():
    return "Hello SSE!"

@app.route('/stream')
def stream():
    event_name = 'custom_event'
    event_data = {'message': 'Hello, world!'}
    event_id = '1'

    sse.publish(event_data, type=event_name, id=event_id, json_format=True)
    return sse.stream()

在上面的代码中，我们使用 sse.publish() 函数发送一个自定义事件。通过参数 type、id 和 json_format，我们可以自由地设置事件的名称和数据格式。