深入理解SSE：构建实时数据推送的前后端解决方案

原创

Front_Yue

发布于 2025-06-30 23:03:45

1.5K00

代码可运行

文章被收录于专栏：码艺坊码艺坊

运行总次数：0

代码可运行

在现代Web应用中，实时数据推送已成为提升用户体验的关键技术之一。本文将深入探讨Server-Sent Events (SSE) 技术，从原理到实践，带你全面掌握如何使用SSE实现前后端实时通信。

一、SSE技术原理

1.1 什么是SSE？

Server-Sent Events (SSE) 是一种基于HTTP的服务器向客户端推送数据的技术。与WebSocket不同，SSE是单向的，仅支持服务器向客户端推送数据，但实现简单，且天然支持断线重连。

1.2 SSE的核心特点

单向通信：仅支持服务器向客户端推送数据
基于HTTP：使用标准HTTP协议，无需额外端口
自动重连：内置重连机制，连接断开后自动尝试重新连接
轻量级：相比WebSocket，协议更简单，开销更小
文本数据：主要传输文本数据，适合JSON等结构化数据

1.3 SSE与WebSocket对比

特性	SSE	WebSocket
通信方向	单向(服务器→客户端)	双向(全双工)
协议	HTTP	独立协议(ws/wss)
数据格式	文本	二进制或文本
复杂度	简单	较复杂
浏览器支持	广泛	广泛
适用场景	实时通知、日志流等单向场景	实时聊天、游戏等双向交互场景

二、SSE工作流程

客户端建立连接：通过EventSource API或自定义fetch请求连接到服务器SSE端点
服务器保持连接：服务器保持HTTP连接打开，不立即关闭
数据推送：服务器通过连接发送数据事件，每条数据以data:开头
客户端接收处理：客户端监听事件并处理接收到的数据
连接管理：自动处理连接断开和重连

三、后端实现SSE

虽然问题主要关注前端实现，但了解后端如何实现SSE有助于更好地理解整个工作流程。以下是Node.js Express框架下的简单SSE实现示例：

// 后端SSE实现示例 (Node.js + Express)
app.post('/api/tool/ai/stream-dialogue', (req, res) => {
  // 设置SSE所需的响应头
  res.setHeader('Content-Type', 'text/event-stream');
  res.setHeader('Cache-Control', 'no-cache');
  res.setHeader('Connection', 'keep-alive');
  res.flushHeaders(); // 立即发送响应头

  // 模拟AI处理过程，分批次发送数据
  const sendData = (type, content) => {
    res.write(`data: ${JSON.stringify({ type, content })}\n\n`);
  };

  // 模拟思考过程
  sendData('think', '正在分析您的问题...');
  
  // 模拟逐步生成回答
  const answerParts = ['这是第一部分回答', '这是第二部分回答', '这是最后一部分回答'];
  answerParts.forEach((part, index) => {
    setTimeout(() => {
      sendData('text', part);
      if (index === answerParts.length - 1) {
        // 最后推荐相关内容
        setTimeout(() => {
          sendData('recommended', '您可能还对以下内容感兴趣...');
          // 结束连接
          res.write('data: {"type":"complete"}\n\n');
          res.end();
        }, 500);
      }
    }, 1000 * (index + 1));
  });
});

四、前端实现SSE的两种方式

4.1 使用原生EventSource API

最简单的SSE前端实现方式是使用浏览器原生提供的EventSource API：

// 使用EventSource API实现SSE客户端
export function dialogueStreamSendWithEventSource(params) {
  const { prompt, onMessage, onReasoning, onRecommended, onError, onClose } = params;
  
  // 创建EventSource连接
  const eventSource = new EventSource(`/api/tool/ai/stream-dialogue?prompt=${encodeURIComponent(JSON.stringify(prompt))}`);
  
  // 监听消息事件
  eventSource.addEventListener('message', (event) => {
    try {
      const data = JSON.parse(event.data);
      if (data.type == 'think') {
        onReasoning(data.content);
      } else if (data.type == 'text') {
        onMessage(data.content);
      } else if (data.type == 'recommended') {
        onRecommended(data.content);
      }
    } catch (err) {
      onError(err);
    }
  });
  
  // 监听错误事件
  eventSource.addEventListener('error', (event) => {
    if (eventEvent.readyState === EventSource.CLOSED) {
      onClose && onClose();
    } else {
      onError && onError(new Error('SSE连接错误'));
    }
  });
  
  // 返回关闭方法
  return {
    close: () => {
      eventSource.close();
      onClose && onClose();
    }
  };
}

注意：原生EventSource API有一些限制，如只能使用GET请求，无法自定义请求头等，这可能不满足所有场景需求。

4.2 使用fetch API实现SSE（如问题中的示例）

对于更复杂的场景，如需要POST请求或自定义请求头，可以使用fetch API手动实现SSE客户端，正如问题中提供的代码示例：

export function dialogueStreamSend(params) {
    const { prompt, onMessage, onReasoning, onRecommended, onError, onClose } = params;
    // 创建AbortController实例
    const controller = new AbortController();
    const signal = controller.signal;
    const fetchStream = async () => {
        try {
            const response = await fetch('/api/tool/ai/stream-dialogue', {
                method: 'POST',
                headers: {
                    'Content-Type': 'application/json',
                },
                body: JSON.stringify(prompt),
                signal // 将signal传给fetch
            });
            if (!response.ok) {
                throw new Error(`HTTP错误! 状态码: ${response.status}`);
            }
            const reader = response.body.getReader();
            const decoder = new TextDecoder('utf-8');
            while (true) {         
                const { done, value } = await reader.read();
                if (done) {
                    if (onClose) onClose();
                    break;
                }
                const chunk = decoder.decode(value, { stream: true });
                const lines = chunk.split('\n');
                let jsonData = '';
                lines.forEach(line => {
                    if (line.includes('type')) {
                        jsonData = line.replace(/^data:/, "").trim(); // 去掉 "data: " 前缀并修剪空格
                        // 如果jsonData为空，跳过解析
                        if (!jsonData) {
                            return;
                        }
                        const data = JSON.parse(jsonData)
                        if(data.type == 'think'){
                            onReasoning(data.content)
                        }
                        if(data.type == 'text'){
                            onMessage(data.content)
                        }
                        if(data.type == 'recommended'){
                            onRecommended(data.content)
                        }
                    }
                });
            }
        } catch (err) {
            if(err.name != 'AbortError'){
                onError(err)
            }   
        }
    };

    // 开始请求
    fetchStream();

    // 返回一个对象，提供close方法用于中止请求
    return {
        close: () => {
            controller.abort();
            if (onClose) onClose();
        }
    };
}

五、在Vue组件中使用SSE

下面展示如何在Vue组件中集成上述SSE功能：

<template>
  <div class="chat-container">
    <div class="messages">
      <div v-for="(message, index) in messages" :key="index" 
           :class="['message', message.type]">
        {{ message.content }}
      </div>
    </div>
    <div v-if="loading" class="loading">AI正在思考...</div>
    <div v-if="error" class="error">{{ error.message }}</div>
    <button @click="startDialogue" :disabled="loading">开始对话</button>
    <button @click="stopDialogue" :disabled="!loading">停止对话</button>
  </div>
</template>

<script>
import { dialogueStreamSend } from '@/api/sse';

export default {
  data() {
    return {
      messages: [],
      loading: false,
      error: null,
      streamController: null
    };
  },
  methods: {
    async startDialogue() {
      this.messages = [];
      this.loading = true;
      this.error = null;
      
      // 准备对话提示
      const prompt = {
        // 这里放置你的对话提示数据
        message: "请解释SSE技术原理",
        context: "用户对实时通信技术感兴趣"
      };
      
      // 启动SSE流
      this.streamController = dialogueStreamSend({
        prompt,
        onMessage: (content) => {
          this.messages.push({ type: 'response', content });
        },
        onReasoning: (content) => {
          this.messages.push({ type: 'thinking', content });
        },
        onRecommended: (content) => {
          this.messages.push({ type: 'recommendation', content });
        },
        onError: (err) => {
          this.error = err;
          this.loading = false;
        },
        onClose: () => {
          this.loading = false;
        }
      });
    },
    stopDialogue() {
      if (this.streamController) {
        this.streamController.close();
        this.streamController = null;
      }
    }
  },
  beforeUnmount() {
    // 组件销毁前确保关闭连接
    this.stopDialogue();
  }
};
</script>

<style scoped>
.chat-container {
  max-width: 600px;
  margin: 0 auto;
}
.messages {
  height: 400px;
  overflow-y: auto;
  border: 1px solid #ddd;
  padding: 10px;
  margin-bottom: 10px;
}
.message {
  margin-bottom: 10px;
  padding: 8px;
  border-radius: 4px;
}
.response {
  background-color: #f0f0f0;
}
.thinking {
  background-color: #e6f7ff;
  font-style: italic;
}
.recommendation {
  background-color: #f6ffed;
  border-left: 3px solid #52c41a;
}
.loading, .error {
  padding: 10px;
  margin-bottom: 10px;
}
.error {
  color: red;
}
</style>

六、进阶优化与注意事项

6.1 错误处理与重连机制

虽然SSE内置了自动重连功能，但在某些情况下可能需要自定义重连逻辑：

export function dialogueStreamSendWithRetry(params, maxRetries = 3) {
  let retryCount = 0;
  
  const attemptConnection = () => {
    const controller = new AbortController();
    const signal = controller.signal;
    
    const fetchStream = async () => {
      try {
        // ...原有fetch代码...
      } catch (err) {
        if (err.name != 'AbortError') {
          if (retryCount < maxRetries) {
            retryCount++;
            console.log(`连接失败，尝试第${retryCount}次重连...`);
            setTimeout(attemptConnection, 2000 * retryCount); // 指数退避
          } else {
            onError(err);
          }
        }
      }
    };
    
    fetchStream();
    
    return {
      close: () => {
        controller.abort();
        if (onClose) onClose();
      }
    };
  };
  
  return attemptConnection();
}