Python使用WebSocket和SSE实现服务端消息实时推送

传统HTTP轮询的瓶颈与推送目标

常规HTTP交互遵循请求-响应模式，客户端需主动发起调用。若依赖短轮询获取实时更新，频繁的请求会占用大量带宽并加重服务端负载，同时网络抖动易导致数据滞后。为实现毫秒级状态同步，通常采用长连接技术替代轮询，核心目标是在保持低资源占用的前提下，建立稳定可靠的双向或单向数据通道。

方案一：WebSocket（全双工通信）

WebSocket 是 HTML5 引入的独立于 HTTP 的协议，基于单个 TCP 连接实现全双工通信。其优势在于客户端与服务端可随时主动发送数据，适合聊天室、实时行情等高交互场景。需要注意的是，不同框架对 WebSocket 版本的兼容性存在差异，实际项目中需确认中间件支持情况。

第1步：服务端监听与路由配置

通过第三方库快速搭建基础服务。以下示例演示如何绑定新接入、断开及消息接收回调，并将指定消息返回给对应客户端。

from websocket_server import WebsocketServer

class WSSocketObj:
    def __init__(self, host=None, port=8131):
        self.host = host if host else '127.0.0.1'
        self.port = port

    def new_client(self, client, server):
        print("新客户端接入 ID: %s" % client['id'])
        server.send_message_to_all("Hey all, a new client has joined us")

    def client_left(self, client, server):
        print("客户端 %d 断开" % client['id'])

    def message_received(self, client, server, message):
        print("Client(%d) said: %s" % (client['id'], message))
        server.send_message(client, 'hello, client')

    def run_server(self):
        server = WebsocketServer(self.port, self.host)
        server.set_fn_new_client(self.new_client)
        server.set_fn_client_left(self.client_left)
        server.set_fn_message_received(self.message_received)
        server.run_forever()

if __name__ == '__main__':
    WSSocketObj().run_server()

第2步：客户端建立连接与消息回调

客户端需维护连接生命周期。Python 端借助线程保持前台阻塞运行，JavaScript 端直接使用原生 WebSocket API 处理事件流。连接成功后即可通过 send 方法投递数据，接收到的载荷由 on_message 统一解析。

# Python 客户端示例
import threading
import websocket

class Client:
    def __init__(self, url):
        self.url = url
        self.ws = None
        self.enable = True

    def on_message(self, response):
        self.enable = False
        print(response)

    def on_error(self, error):
        print(error)

    def on_close(self):
        self.enable = True
        print("连接已关闭")

    def on_open(self):
        print('连接已建立')

    def start_func(self):
        while self.enable:
            websocket.enableTrace(True)
            self.ws = websocket.WebSocketApp(
                self.url,
                on_message=self.on_message,
                on_error=self.on_error,
                on_open=self.on_open,
                on_close=self.on_close
            )
            self.ws.run_forever(ping_interval=60, ping_timeout=5)

if __name__ == "__main__":
    cli = Client(url='wss://api.zb.live/websocket')
    t1 = threading.Thread(target=cli.start_func)
    t1.start()

// JavaScript 客户端示例
var ws = new WebSocket("wss://echo.websocket.org");

ws.onopen = function(evt) { 
  console.log("Connection open ..."); 
  ws.send("Hello WebSockets!");
};

ws.onmessage = function(evt) {
  console.log("Received Message: " + evt.data);
  ws.close();
};

ws.onerror = function(evt) {
  console.log("Connection error.");
};

ws.onclose = function(evt) {
  console.log("Connection closed.");
};

方案二：SSE（Server-Sent Events）

SSE 是一种基于标准 HTTP 协议的单向流式传输技术。服务端持续输出带有特定格式的事件流，客户端自动解析。其开发门槛低于 WebSocket，天然复用 HTTP 鉴权与代理缓存，但仅支持服务端向客户端推送。若需双向通信，需额外开辟普通 HTTP 接口。

第1步：Flask服务端构建流式响应

利用 flask.Response 返回 text/event-stream 类型数据。关键在于使用生成器 yield 保持连接不中断，并按规范拼接 id、event、data 字段。

import json
import time
from flask import Flask, request, Response, render_template

app = Flask(__name__)

def get_message():
    time.sleep(1)
    s = time.ctime(time.time())
    return json.dumps(['当前时间：' + s, 'a'], ensure_ascii=False)

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

@app.route('/stream')
def stream():
    user_id = request.args.get('user_id')
    print(user_id)
    
    def eventStream():
        msg_id = 0
        while True:
            msg_id += 1
            # 等待数据就绪后推送，严格遵循SSE格式
            yield f'id: {msg_id}\nevent: add\ndata: {get_message()}\n\n'
            time.sleep(1)

    return Response(eventStream(), mimetype="text/event-stream")

if __name__ == '__main__':
    app.run(debug=True)

第2步：客户端解析事件流

Python 可通过流式请求配合解析库读取，浏览器端推荐使用原生 EventSource。该对象内置状态管理与自动重连逻辑，开发者仅需挂载 onopen、onmessage 与 onerror 处理器。

# Python SSE 客户端（方法一：专用库）
def sse_client_with_lib():
    import requests
    from sseclient import SSEClient
    client = requests.post(url, json=data, stream=True, headers={'Accept': 'text/event-stream'})
    parser = SSEClient(client)
    for event in parser.events():
        print(event.data)

# Python SSE 客户端（方法二：原生迭代）
def sse_client_raw():
    import requests
    headers = {"Accept": "text/event-stream"}
    r = requests.post(url, headers=headers, json=data, stream=True)
    r.encoding = 'utf-8'
    for chunk in r.iter_content(chunk_size=None, decode_unicode=True):
        print("Received:", chunk)

// JavaScript 客户端示例
if ('EventSource' in window) {
  const source = new EventSource('http://localhost:8000/stream');

  source.addEventListener('open', () => {
    console.log('SSE 连接已开启');
  });

  source.addEventListener('add', e => {
    document.body.append(document.createElement('p')).textContent = e.data;
  });

  source.addEventListener('error', e => {
    console.log('连接异常或断开:', e.target.readyState);
  });
}

SSE协议规范与可靠性保障

SSE 报文必须为 UTF-8 编码文本，每条消息以 \n\n 作为结束定界符。规范定义了四个核心字段：id 用于标识唯一事件序列，event 声明事件类型，data 承载有效载荷，retry 设定客户端静默后的重连间隔（毫秒）。此外，以冒号开头的内容被视为注释，常用于维持长连接心跳，不会触发前端事件回调。

为保证数据完整性，务必在每条推送中附带递增的 id 字段。当网络波动触发 error 事件时，客户端会自动发起重连，并在请求头携带 Last-Event-ID，服务端据此跳过已送达片段。生产环境中可结合自定义控制事件主动释放连接，避免资源闲置：

function connectSSE() {
  if (!window.EventSource) return;
  const source = new EventSource('http://localhost:2000');
  let reconnectTimeout;

  source.addEventListener('open', () => {
    clearTimeout(reconnectTimeout);
  });

  source.addEventListener('pause', e => {
    source.close();
    const targetTime = Number(e.data);
    const delay = targetTime - Date.now();
    reconnectTimeout = setTimeout(connectSSE, Math.max(delay, 0));
  });
}

简要总结

WebSocket 凭借全双工特性适合高频双向交互，而 SSE 依托标准 HTTP 协议更易于集成与扩展，适合仪表盘、日志监控等单向推送场景。在实际选型时，应综合考量网络代理兼容性、客户端生态支持以及服务端并发模型。合理运用字段标识与心跳机制，可有效提升长连接的稳定性与资源利用率。

以上就是基于Python实现WebSocket与SSE服务端消息推送的详细内容，更多关于实时通信架构设计与长连接优化实践的资料请关注本站其它相关文章！