如何在Python（Flask）中实现SSE服务端与客户端？

服务端（Flask）实现 #

Flask实现SSE的核心是返回流式响应，严格遵循SSE协议规范设置响应头，同时按格式推送事件数据。

基础示例代码 #

from flask import Flask, Response
import time

app = Flask(__name__)

# 模拟事件生成逻辑，实际场景可替换为业务触发事件（如数据库变更、消息队列通知）
def generate_sse_events():
    count = 0
    while True:
        # SSE标准格式：data: 消息内容\n\n，可按需添加id: 事件ID\n或event: 事件类型\n
        yield f"data: 第{count}条事件 - {time.strftime('%Y-%m-%d %H:%M:%S')}\n\n"
        count += 1
        time.sleep(3)  # 模拟事件间隔，实际根据业务需求调整

@app.route('/sse')
def sse_subscribe():
    return Response(
        generate_sse_events(),
        mimetype='text/event-stream',
        headers={
            'Cache-Control': 'no-cache',
            'Connection': 'keep-alive'
        }
    )

if __name__ == '__main__':
    app.run(debug=True, threaded=True)  # 开启多线程，避免单线程阻塞其他请求

关键要点 #

• mimetype='text/event-stream'：必须设置，告知客户端这是SSE流类型响应
• Cache-Control: no-cache：防止客户端缓存响应内容，确保实时接收新事件
• threaded=True：Flask默认单线程，开启多线程才能同时处理多个客户端订阅请求
• 若需触发特定业务事件（而非定时推送），可结合Redis Pub/Sub等消息队列，在事件产生时向所有订阅客户端推送数据。

Python客户端实现 #

使用requests库发起流式GET请求，逐行解析SSE格式的响应内容。

示例代码 #

import requests

def sse_client():
    url = 'http://localhost:5000/sse'
    with requests.get(url, stream=True) as response:
        response.raise_for_status()
        # 逐行遍历响应流，过滤空行并解析SSE消息
        for line in response.iter_lines(decode_unicode=True):
            if line.startswith('data:'):
                message_content = line[5:].strip()
                print(f"收到事件: {message_content}")

if __name__ == '__main__':
    sse_client()

关键要点 #

• stream=True：保持长连接，逐行接收服务器推送的事件
• 仅处理以data:开头的行，符合SSE协议的消息格式
• 如需处理事件类型或ID，可扩展解析逻辑（如匹配event:或id:开头的行）

内容的提问来源于stack exchange，提问作者Matteo Bellin