SSE协议

顾翔

发布于 2025-10-11 11:22:49

2870

文章被收录于专栏：啄木鸟软件测试啄木鸟软件测试

1.什么是 SSE？

SSE（Server-Sent Events）是一种允许服务器主动向客户端（通常是浏览器）推送数据的 Web 技术。它建立在标准的 HTTP 协议之上，是 HTML5 规范的一部分。

与传统的客户端轮询（Polling）或长轮询（Long-Polling）相比，SSE 提供了一种更高效、更简单的服务器到客户端的单向通信通道。

2.核心特点

1)单向通信：数据流只能从服务器发送到客户端。客户端无法通过 SSE 连接向服务器发送数据（但可以通过其他方式，如fetch或 XHR）。

2)基于HTTP/HTTPS：SSE 使用普通的HTTP 协议，不需要特殊的协议或端口，因此更容易与现有基础设施集成，也更容易通过防火墙。

3)文本数据：默认情况下，SSE 设计用于传输 UTF-8 编码的文本数据。虽然可以通过技巧传输二进制数据，但并非其初衷。

4)内置重连机制：协议本身支持自动重连。如果连接断开，浏览器会自动尝试重新建立连接。

5)简单易用：在浏览器端有原生 JavaScript API（EventSource），使用起来非常简单。

3.工作原理

1)客户端发起请求：客户端（浏览器）使用EventSource对象向一个特定的服务器端点发起一个常规的 HTTP 请求。

2)服务器保持连接打开：服务器接收到请求后，将响应头Content-Type设置为text/event-stream，并保持这个 HTTP 连接处于打开状态（Keep-Alive）。

3)服务器持续发送数据：服务器可以随时通过这个持久连接向客户端发送数据。数据的格式遵循简单的规范（见下文）。

4)客户端处理数据：客户端的EventSource对象会监听来自服务器的消息，并在收到时触发事件（如onmessage）来处理数据。

5)连接终止或重连：连接会一直保持，直到一方（服务器或客户端）主动关闭它。如果连接意外中断，EventSource会自动尝试重新连接。

4. 数据格式规范

服务器发送的响应体不是任意文本，而必须遵循特定的格式。每条消息由一系列字段组成，每个字段由fieldname成，以换行符结束。消息之间用一个空行（即两个换行符\n\n）: value构分隔。

常用的字段有四个：

1. data：消息的数据内容。一行数据可以写为data: some text\n。如果数据很长，可以分成多行，每行都以data:开头。

text

data: This is the first line.
data: This is the second line.

客户端接收到时，会将多行data字段的值用换行符连接起来，形成"This is the first line.\nThis is the second line."。

2. event：事件类型标识符。这是一个字符串，用于区分不同种类的消息。客户端可以根据这个类型来监听不同的事件（而不仅仅是通用的 message 事件）。

text

event: userjoined
data: {"username": "Alice"}

3. id：消息的唯一标识符（ID）。通常是自增的数字或字符串。浏览器在重连时，会通过 HTTP 头Last-Event-ID将这个最后收到的 ID 发送给服务器，帮助服务器知道客户端从哪里断开的，从而发送遗漏的消息。

text

id: 12345
data: Something happened

4. retry：指定浏览器在连接断开后重连的等待时间（单位：毫秒）。服务器可以建议客户端下次重连的间隔。

text

retry: 10000

这告诉浏览器：“如果连接断了，10秒后再尝试重连。”

一个完整的 SSE 响应流示例：

http

HTTP/1.1200OK

Content-Type: text/event-stream

Cache-Control: no-cache

Connection: keep-alive

: This is a comment line (optional)

event: status

data: System is starting up...

id: 1

data: Hello,

data: World!

id: 2

event: update

data: {"value": 42}

id: 3

5.浏览器端 API：EventSource

在浏览器中，使用EventSource对象来处理 SSE 连接。

基本用法：

javascript

// 1. 创建 EventSource 对象，连接到服务器端点

const eventSource =newEventSource('/api/events');

// 2. 监听通用的消息事件（没有指定 event 类型的消息）

eventSource.onmessage=function(event){

console.log('Generic message:', event.data);

};

// 3. 监听特定自定义事件（对应服务器消息中的 event 字段）

eventSource.addEventListener('userjoined',function(event){

const userData =JSON.parse(event.data);

console.log('User joined:', userData.username);

});

// 4. 监听错误事件

eventSource.onerror=function(error){

console.error('EventSource error:', error);

// eventSource.readyState 属性会表明连接状态

// 0: CONNECTING, 1: OPEN, 2: CLOSED

};

关闭连接：

javascript

eventSource.close();// 客户端主动终止连接

6.优点与缺点

优点：

l 简单：API 简单易懂，无需额外的库或复杂的协议。

l 高效：相比频繁的轮询，减少了不必要的HTTP 请求开销和头部信息。

l 自动重连：内置的重连机制提高了连接的可靠性。

l 标准支持：是 Web 标准的一部分，得到所有现代浏览器的良好支持。

缺点：

l 单向通信：只能服务器推数据到客户端。如果需要双向通信，WebSocket 是更好的选择。

l 文本协议：原生不支持二进制数据流，主要传输文本（如 JSON）。

l 连接数限制：浏览器对到同一域下的最大HTTP 连接数有限制（通常是 6 个），SSE 和 Ajax 等会共享这个限制。

l 不支持旧版浏览器：如 Internet Explorer 完全不支持EventSource。

7．与 WebSocket 的对比

特性	SSE (Server-Sent Events)	WebSocket
通信方向	单向(服务器 -> 客户端)	双向(全双工)
协议	HTTP	独立的 WS/WSS 协议 (基于 TCP)
数据格式	文本(UTF-8)	文本和二进制帧
复杂度	简单	相对复杂
自动重连	内置支持	需要手动实现

总结

SSE是一种非常优雅和实用的技术，专门为服务器向客户端实时推送数据的场景而设计。它利用简单的 HTTP 协议和文本格式，提供了高效、可靠且易于使用的单向通信通道。如果你的应用主要是由服务器驱动向客户端发送更新（如股票报价、新闻源、实时监控仪表盘），SSE 通常是一个比 WebSocket 更简单、更合适的选择。

一个案例

现在，我们用flask模拟一个SSH协议

服务器端代码

import json
import time
from flask import Flask,request-
from flask import Response
from flask import render_template
app= Flask(__name__)
def get_message():
    """等待数据就绪"""
    time.sleep(1)
    s = time.ctime(time.time())
    return json.dumps(['当前时间:'+s ,'sse server'],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():
        id = 0
        while True:
            id +=1
            
#等待数据可用，然后处理
            yield 'id: {}\nevent: add\ndata: {}\n\n'.format(id,get_message())
    return Response(eventStream(), mimetype="text/event-stream")
if __name__=='__main__':
  app.run()

建立templates目录，在templates下建立index.html

<!DOCTYPE html>
<html>
<head>
    <title>SSE 示例</title>
    <style>
        body {
            font-family: Arial, sans-serif;
            max-width: 800px;
            margin: 0 auto;
            padding: 20px;
        }
        
#events
 {
            border: 1px solid 
#ddd
;
            padding: 15px;
            height: 300px;
            overflow-y: auto;
            margin-top: 20px;
        }
        .event {
            margin-bottom: 10px;
            padding: 10px;
            border-left: 3px solid 
#4CAF50
;
            background-color: 
#f9f9f9
;
        }
    </style>
</head>
<body>
    <h1>SSE 服务器示例</h1>
    <p>打开浏览器控制台查看SSE事件，或者使用以下代码测试：</p>
    <button onclick="connectSSE()">连接SSE</button>
    <button onclick="disconnectSSE()">断开连接</button>
    <div id="events">
        <p>事件将显示在这里...</p>
    </div>
    <script>
        let eventSource;
        function connectSSE() {
            eventSource = new EventSource('/stream?user_id=test_user');
            eventSource.onmessage = function(event) {
                const data = JSON.parse(event.data);
                addEvent('消息', data, event.lastEventId);
            };
            eventSource.onopen = function() {
                addEvent('连接', 'SSE连接已建立');
            };
            eventSource.onerror = function() {
                addEvent('错误', '连接错误或已关闭');
            };
            addEvent('状态', '开始接收服务器事件');
        }
        function disconnectSSE() {
            if (eventSource) {
                eventSource.close();
                addEvent('状态', '已停止接收服务器事件');
            }
        }
        function addEvent(type, data, id = '') {
            const eventsDiv = document.getElementById('events');
            const eventElement = document.createElement('div');
            eventElement.className = 'event';
            let content = `<strong>${type}</strong>`;
            if (id) content += ` (ID: ${id})`;
            content += `<br>`;
            if (Array.isArray(data)) {
                content += data.join(' | ');
            } else {
                content += data;
            }
            eventElement.innerHTML = content;
            eventsDiv.appendChild(eventElement);
            // 自动滚动到底部
            eventsDiv.scrollTop = eventsDiv.scrollHeight;
        }
    </script>
</body>
</html>

运行。终端显示

======= RESTART: C:\Users\xiang\Desktop\SSE_1.py =======
* Serving Flask app 'SSE_1'
* Debug mode: off
[31m[1mWARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead.[0m
* Running on http://127.0.0.1:5000
[33mPress CTRL+C to quit[0m

打开浏览器，输入127.0.0.1:5000/stream

每隔1秒，输出时间。

打开浏览器，输入127.0.0.1:5000

假数据，作用，后端运行返回码不是500.

客户端代码

import requests
import json
def test_sse_server():
    url = "http://127.0.0.1:5000/stream?user_id=test_user"
    headers = {'Accept': 'text/event-stream'}
    try:
        response = requests.get(url, stream=True, headers=headers)
        if response.status_code == 200:
            print("成功连接到SSE服务器")
            print("开始接收事件 (按Ctrl+C停止)...")
            for line in response.iter_lines(decode_unicode=True):
                if line:
                    print(f"收到: {line}")
        else:
            print(f"连接失败，状态码: {response.status_code}")
    except KeyboardInterrupt:
        print("\n停止接收SSE流.")
    except Exception as e:
        print(f"连接失败: {e}")
if __name__ == "__main__":
test_sse_server()

运行

测试代码

import requests
def test_sse_connection():
    """测试SSE连接"""
    response = requests.get('http://localhost:5000/stream', stream=True, headers={'Accept': 'text/event-stream'})
    assert response.status_code == 200
    assert response.headers['Content-Type'] == 'text/event-stream; charset=utf-8'
def test_sse_data():
    """测试SSE数据"""
    response = requests.get('http://localhost:5000/stream', stream=True, headers={'Accept': 'text/event-stream'})
    # 读取一些数据来验证内容
    data_found = False
    line_count = 0
    for line in response.iter_lines(decode_unicode=True):
        if line and line.startswith('data:'):
            data = line[5:].strip()
            if "sse" in data.lower() or "当前时间" in data:
                data_found = True
                break
        line_count += 1
        if line_count >= 10:  # 安全限制
            break
    assert data_found, "未找到预期的数据内容"

运行测试代码

C:\Users\xiang\Desktop>pytest -sv TestSSE.py
=================test session starts =================
platform win32 -- Python 3.12.3, pytest-8.4.1, pluggy-1.6.0 -- C:\Users\xiang\AppData\Local\Programs\Python\Python312\python.exe
cachedir: .pytest_cache
metadata: {'Python': '3.12.3', 'Platform': 'Windows-11-10.0.26100-SP0', 'Packages': {'pytest': '8.4.1', 'pluggy': '1.6.0'}, 'Plugins': {'allure-pytest': '2.13.5', 'anyio': '4.3.0', 'pyfakefs': '5.8.0', 'base-url': '2.1.0', 'cov': '6.2.1', 'html': '3.2.0', 'metadata': '3.1.1', 'playwright': '0.5.0'}, 'JAVA_HOME': 'C:\\Tools\\jdk-17.0.15', 'Base URL': ''}
rootdir: C:\Users\xiang\Desktop
plugins: allure-pytest-2.13.5, anyio-4.3.0, pyfakefs-5.8.0, base-url-2.1.0, cov-6.2.1, html-3.2.0, metadata-3.1.1, playwright-0.5.0
collected 2 items
TestSSE.py::test_sse_connection PASSED
TestSSE.py::test_sse_data PASSED
=========== 2 passed in 6.50s ===========

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2025-09-10，如有侵权请联系 cloudcommunity@tencent.com 删除

sse