当前位置：首页 > article >正文

SSE 流式场景应用及方案总结

article 2024/12/26 16:16:30

文章目录

- SSE 流式场景
- 前端流式解码应用
- - JavaScript中的EventSource API
  - SSE 与 Axios应用
  - SSE 与 fetch 应用
  - fetch 自己的流式响应
  - 代码解释

SSE 流式场景

股票行情实时推送
- 背景：在金融领域，股票价格是实时变化的。投资者和交易员需要及时获取最新的股票行情信息来做出交易决策。传统的请求 - 响应模式（如通过定期刷新网页或轮询API）可能会导致数据延迟或网络资源浪费。
- 实现细节：服务器端（如证券交易所的数据中心）可以使用SSE（Server - Sent Events）来实时推送股票价格更新。当有新的股票价格数据产生时，服务器会立即将数据通过SSE流发送给已连接的客户端（如股票交易软件或金融新闻网站的前端界面）。例如，假设一个股票交易应用，当用户关注了某几只股票后，服务器会针对该用户建立一个SSE连接，一旦这些股票的价格发生变化，服务器就会发送类似如下的数据格式：

data:{"symbol":"AAPL","price":180.50,"change": + 0.25}

优势：客户端可以实时接收最新的股票价格变化，无需频繁地向服务器发送请求。这大大提高了数据的及时性，并且减少了网络带宽的占用，因为数据是在有更新时才发送，而不是周期性地请求可能没有变化的数据。

实时日志监控系统
- 背景：在软件开发和运维中，需要实时查看应用程序的日志来快速发现和解决问题。对于分布式系统或大型服务器集群，日志可能会分散在多个节点上，传统的查看日志方式（如手动登录到每个服务器查看日志文件）效率低下。
- 实现细节：应用程序的日志收集工具可以将新产生的日志条目通过SSE发送到监控控制台。例如，一个Web服务器集群的日志监控系统，当有新的访问请求被处理时，日志信息（如访问时间、请求的URL、响应状态码等）会被收集并通过SSE发送到监控界面。数据可能以这样的格式发送：

data:{"timestamp":"2024 - 12 - 25T10:30:00Z","level":"INFO",
"message":"User accessed /index.html","source":"web - server - 001"}

优势：运维人员可以在一个集中的监控控制台实时查看所有服务器的日志更新，及时发现异常情况，如频繁的404错误或服务器性能下降的迹象。这种实时性有助于快速响应问题，减少系统停机时间。

实时体育比赛比分推送
- 背景：体育爱好者希望能够在比赛进行过程中实时获取比分更新和比赛事件（如进球、红黄牌等）。体育媒体网站和移动应用需要一种高效的方式来向用户推送这些实时信息。
- 实现细节：体育数据供应商（如专业的体育数据公司）可以使用SSE来将比赛数据发送给媒体平台。以一场足球比赛为例，每当有进球、角球、犯规等事件发生时，服务器会发送如下格式的数据：

data:{"matchId":"12345","event":"Goal","team":"Home",
"player":"Player A","time": "30'"}

优势：用户可以在体育媒体应用或网站上实时看到比分和比赛事件的更新，增强了用户体验。同时，媒体平台可以轻松地集成这种实时数据推送功能，而无需复杂的轮询机制来获取最新比分。

前端流式解码应用

JavaScript中的EventSource API

背景知识：
- 在前端JavaScript中，当后端使用SSE（Server - Sent Events）进行流式输出时，可以使用EventSource对象来接收和处理这些流数据。EventSource提供了一种简单的方式来监听服务器发送的事件。
具体实例方法：
- 首先，在HTML文件中创建一个用于显示数据的元素，例如一个<div>元素：

<!DOCTYPE html>
<html>
<head>
    <title>SSE Example</title>
</head>
<body>
    <div id="data-display"></div>
    <script>
        // 创建一个EventSource对象，连接到后端的SSE端点
      const eventSource = new EventSource('http://your-backend-url/sse-endpoint');
        const dataDisplay = document.getElementById('data - display');
        // 监听message事件，这是SSE默认发送数据的事件类型
        eventSource.addEventListener('message', function (event) {
            // 将接收到的数据添加到页面的div元素中
            dataDisplay.innerHTML += '<p>' + event.data + '</p>';
        });
        // 当连接关闭时（例如，服务器端主动关闭连接或网络问题）
        eventSource.addEventListener('error', function (event) {
            if (event.target.readyState === EventSource.CLOSED) {
                console.log('Connection was closed.');
            } else {
                console.log('There was an error with the connection.');
            }
        });
    </script>
</body>
</html>

解释说明：
- new EventSource('http://your - backend - url/sse - endpoint')：创建一个EventSource对象，其中http://your - backend - url/sse - endpoint是后端提供SSE服务的URL端点。这个端点会持续发送数据。
- eventSource.addEventListener('message', function (event) {... })：监听message事件，当后端通过SSE发送数据时，会触发这个事件。event.data包含了后端发送的数据，这里将数据以<p>标签包裹后添加到id为data - display的<div>元素中，从而在页面上显示出来。
- eventSource.addEventListener('error', function (event) {... })：监听连接错误事件。如果连接被关闭（event.target.readyState === EventSource.CLOSED），会在控制台打印Connection was closed.；如果是其他连接错误，会打印There was an error with the connection.。

SSE 与 Axios应用

背景知识：
- Axios是一个广泛使用的HTTP客户端库，它也可以处理一些类似流式的响应，虽然它不是专门为SSE设计的，但在某些场景下可以用来接收和处理后端持续发送的数据。
具体实例方法：
- 首先，确保已经在项目中安装了Axios（可以通过npm install axios安装）。以下是一个示例代码：

<!DOCTYPE html>
<html>
<head>
    <title>Axios Stream Example</title>
</head>
<body>
    <div id="data - display"></div>
    <script src="https://unpkg.com/axios/dist/axios.min.js"></script>
    <script>
        const dataDisplay = document.getElementById('data - display');
        const source = new EventSource('http://your-backend-url/sse-endpoint');
        // 创建一个Axios CancelToken源，用于取消请求（如果需要）
        const CancelToken = axios.CancelToken;
        let cancel;
        source.addEventListener('message', function (event) {
            // 使用Axios发送一个简单的POST请求，这里只是示例，可以根据实际情况修改
            axios.post('http://your-backend-url/process-data',{data:event.data },{
                cancelToken: new CancelToken(function executor(c) {
                    cancel = c;
                })
            }).then(function (response) {
                dataDisplay.innerHTML += '<p>' + response.data + '</p>';
            }).catch(function (error) {
                if (axios.isCancel(error)) {
                    console.log('Request canceled', error.message);
                } else {
                    console.log('There was an error:', error);
                }
            });
        });
        // 当连接关闭时，取消所有未完成的Axios请求（如果有）
        source.addEventListener('error', function (event) {
            if (cancel) {
                cancel();
            }
        });
    </script>
</body>
</html>

解释说明：
- 这段代码结合了EventSource和Axios。首先，当EventSource接收到message事件时，会获取后端发送的数据（event.data）。
- 然后使用Axios发送一个POST请求到http://your - backend - url/process - data，并将接收到的数据作为请求数据发送。通过cancelToken可以在需要时取消请求。
- 如果Axios请求成功，会将响应数据（response.data）以<p>标签包裹后添加到id为data - display的<div>元素中。如果出现错误，会根据错误类型（是请求被取消还是其他错误）在控制台打印相应的信息。当EventSource连接关闭时，会取消所有未完成的Axios请求。这种方式在需要对后端发送的数据进行进一步处理（如发送到另一个后端端点）时比较有用。## 技术名词解释

SSE 与 fetch 应用

实时数据更新与初始数据获取结合
- 场景描述
  - 假设一个社交媒体应用，用户打开页面时，需要先获取最新的一些帖子信息来填充页面，同时还希望能够实时收到新帖子的推送。这时可以结合fetch和SSE来实现这个功能。
- 具体实现
  - 首先，使用fetch获取初始的帖子数据：

fetch('https://your - api - url/posts')
   .then(response => response.json())
   .then(data => {
        // 处理初始数据，例如将帖子显示在页面上
        data.forEach(post => {
            const postElement = document.createElement('div');
            postElement.textContent = post.content;
            document.body.appendChild(postElement);
        });
    });

 - 然后，使用`SSE`接收新帖子的实时推送：

const eventSource = new EventSource('https://your - api - url/posts - stream');
eventSource.addEventListener('message', function (event) {
    const newPost = JSON.parse(event.data);
    const newPostElement = document.createElement('div');
    newPostElement.textContent = newPost.content;
    document.body.appendChild(newPostElement);
});

解释说明
- 对于初始数据，fetch请求https://your - api - url/posts端点获取帖子列表。当获取到数据后（通过.then(response => response.json())将响应解析为JSON格式），遍历数据并为每个帖子创建一个<div>元素，将帖子内容添加到元素中，最后将元素添加到页面的body部分。
- 对于实时数据推送，创建EventSource对象连接到https://your - api - url/posts - stream端点。当接收到message事件（即有新帖子推送）时，将接收到的数据（event.data）解析为JSON格式（JSON.parse(event.data)），创建新的<div>元素并添加到页面，这样用户就能实时看到新帖子。

数据缓存更新与实时通知
- 场景描述
  - 在一个电商应用中，用户可能已经获取并缓存了一些商品信息（如价格、库存等）。当商品信息在后端发生变化时，希望通过SSE实时通知前端，同时前端可以使用fetch来更新缓存中的数据。
- 具体实现
  - 首先，假设已经有一个函数用于缓存商品数据：

let productCache = {};
async function cacheProductData(productId) {
    const response = await fetch(`https://your-api-url/products/${productId}`);
    const data = await response.json();
    productCache[productId] = data;
    return data;
}

然后，使用SSE接收商品信息更新的通知：

const eventSource = new EventSource('https://your-api-url/product-updates');
eventSource.addEventListener('message', async function (event) {
    const updateData = JSON.parse(event.data);
    const productId = updateData.productId;
    // 使用fetch更新缓存中的商品数据
    const updatedData = await cacheProductData(productId);
    // 可以在这里进行页面更新，比如更新商品价格显示等
    const priceElement = document.getElementById(`price - ${productId}`);
    if (priceElement) {
        priceElement.textContent = updatedData.price;
    }
});

解释说明
- 对于数据缓存，cacheProductData函数是一个异步函数，它使用fetch请求商品数据，将响应解析为JSON格式后，将数据存储到productCache对象中，并返回数据。这个函数可以在页面加载时或者用户查看商品详情时调用，用于缓存商品信息。
- 当SSE接收到商品信息更新的通知（message事件）时，先将通知数据（event.data）解析为JSON格式。获取商品ID后，调用cacheProductData函数更新缓存中的商品数据。之后，可以根据更新的数据来更新页面元素，例如通过id找到对应的商品价格元素（price - ${productId}），并更新其显示的价格。

fetch 自己的流式响应

在JavaScript中，fetch本身可以处理流式响应，通过body属性获取响应体的ReadableStream对象，然后对其进行操作。以下是一个示例，展示如何使用fetch接收流式输出并逐块处理数据：

<!DOCTYPE html>
<html lang="en">

<head>
    <meta charset="UTF - 8">
    <meta name="viewport" content="width=device-width, initial - scale = 1.0">
    <title>Fetch Stream Example</title>
</head>

<body>
    <div id="output"></div>
    <script>
        async function fetchStream() {
            const response = await fetch('your - streaming - endpoint');
            if (!response.ok) {
                throw new Error(`HTTP error! status: ${response.status}`);
            }
            const reader = response.body.getReader();
            const decoder = new TextDecoder('utf - 8');
            let result = '';
            const outputDiv = document.getElementById('output');
            while (true) {
                const { done, value } = await reader.read();
                if (done) {
                    break;
                }
                const chunk = decoder.decode(value, { stream: true });
                result += chunk;
                // 这里简单将每块数据追加到页面上，实际应用中可根据需求处理
                outputDiv.textContent += chunk;
            }
            // 处理最后的数据
            result += decoder.decode();
            console.log('Final result:', result);
        }
        fetchStream().catch(console.error);
    </script>
</body>

</html>

代码解释

发起请求：
- const response = await fetch('your - streaming - endpoint');：使用fetch发起一个HTTP请求到指定的流式输出端点。
- if (!response.ok) { throw new Error(HTTP error! status: ${response.status}); }：检查响应状态，如果不是2xx状态码，抛出错误。
获取读取器和解码器：
- const reader = response.body.getReader();：从响应体的body属性获取一个ReadableStreamDefaultReader对象，用于读取流数据。
- const decoder = new TextDecoder('utf - 8');：创建一个TextDecoder对象，用于将读取的二进制数据解码为UTF - 8格式的文本。
循环读取数据：
- while (true) {... }：进入一个无限循环，用于逐块读取数据。
- const { done, value } = await reader.read();：调用reader.read()方法读取一块数据，done表示是否读取到流的末尾，value是读取到的数据块（Uint8Array类型）。
- const chunk = decoder.decode(value, { stream: true });：使用decoder.decode方法将数据块解码为文本，{ stream: true }选项表示这不是流的最后一块数据，解码器会保留内部状态以处理后续数据。
- result += chunk;：将解码后的文本块添加到result变量中。
- outputDiv.textContent += chunk;：将每块解码后的文本追加到页面的div元素中，实时显示数据。
处理最后的数据：
- result += decoder.decode();：在读取完所有数据块后，调用decoder.decode()方法（不传递参数）处理解码器中剩余的数据。
- console.log('Final result:', result);：将最终结果打印到控制台。