1. 什么是 SSE？

• 定义：SSE（Server-Sent Events，服务器发送事件）是一种基于 HTTP 协议的单向实时通信技术，允许服务器持续向客户端推送数据，客户端无需重复发起请求。
• 特点：
  • 单向通信：仅服务器向客户端推送数据
  • 长连接：一次连接持续传输，减少握手开销
  • 文本协议：数据以 UTF-8 格式传输，遵循特定格式规范
  • 自动重连：客户端断开后会自动尝试重连（默认机制）

2. 为什么需要 EventSourcePolyfill？

• 原生 EventSource 局限性：
  • 浏览器兼容性差：不支持 IE 全版本、低版本 Safari/Chrome
  • 不支持自定义请求头：无法携带认证 Token、Cookie 等关键信息
  • 重连机制不灵活：缺乏自定义超时和重连策略配置
• EventSourcePolyfill 优势：
  • 完全兼容原生 EventSource API，无缝替换
  • 支持自定义请求头（如 Authorization）
  • 可配置心跳超时、重连策略
  • 兼容所有现代浏览器及 IE 10+

3. 流式数据传输的应用场景

• AI 大模型对话（逐字生成响应）
• 实时日志监控（后端日志实时推送）
• 股票 / 行情数据实时更新
• 新闻 / 通知实时推送
• 在线协作工具（实时同步状态）

三、环境准备

1. 安装依赖

（1）npm 安装

bash

运行

# 核心依赖
npm install eventsource-polyfill --save

# TypeScript 项目需安装类型声明
npm install @types/eventsource-polyfill --save-dev

（2）CDN 引入（非工程化项目）

html

预览

<script src="https://cdn.jsdelivr.net/npm/eventsource-polyfill@1.0.31/dist/eventsource.min.js"></script>

2. 浏览器兼容性验证

浏览器	支持版本	备注
Chrome	4+	原生支持，polyfill 兼容
Firefox	6+	原生支持，polyfill 兼容
Safari	5+	低版本需 polyfill 支持
Edge	12+	原生支持，polyfill 兼容
IE	10+	仅 polyfill 支持
移动端浏览器	主流版本均支持	需 polyfill 适配低版本

四、基础使用教程

1. 核心 API 详解

（1）构造函数

javascript

运行

new EventSourcePolyfill(url, options);

• url：SSE 服务端接口地址（必填）
• options：配置对象（可选）

配置项	类型	说明	默认值
headers	Object	自定义请求头（如认证 Token）	{}
heartbeatTimeout	Number	心跳超时时间（毫秒），超时后重连	45000
withCredentials	Boolean	跨域时是否携带 Cookie	false
reconnectInterval	Number	重连间隔时间（毫秒）	1000

（2）核心方法

方法名	说明
close()	关闭 SSE 连接，停止接收数据
addEventListener(type, callback)	监听指定类型事件
removeEventListener(type, callback)	移除事件监听

（3）核心事件

事件名	触发时机
message	接收服务器默认格式数据（无指定事件名）
open	连接建立成功时
error	连接出错或断开时
自定义事件	服务器指定 event 字段时（如 notification）

2. 最小化示例（原生 JS）

（1）前端代码

javascript

运行

// 引入 polyfill（工程化项目）
import EventSource from 'eventsource-polyfill';

// 初始化 SSE 连接
const es = new EventSourcePolyfill('http://localhost:3001/sse', {
  headers: {
    'Authorization': 'Bearer your-auth-token' // 携带认证 Token
  },
  heartbeatTimeout: 60000 // 60秒心跳超时
});

// 连接建立成功
es.addEventListener('open', () => {
  console.log('SSE 连接已建立');
});

// 接收默认格式数据
es.addEventListener('message', (e) => {
  console.log('收到数据：', JSON.parse(e.data));
});

// 接收自定义事件数据
es.addEventListener('notification', (e) => {
  console.log('收到通知：', JSON.parse(e.data));
});

// 连接错误处理
es.addEventListener('error', (err) => {
  console.error('SSE 连接错误：', err);
  // 可自定义重连逻辑
  if (err.readyState === EventSource.CLOSED) {
    console.log('连接已关闭，准备重连...');
  }
});

// 手动关闭连接（如页面卸载时）
window.addEventListener('beforeunload', () => {
  es.close();
});

（2）服务端代码（Node.js/Express）

javascript

运行

const express = require('express');
const cors = require('cors');
const app = express();

// 跨域配置（SSE 必须允许跨域）
app.use(cors({
  origin: 'http://localhost:3000', // 前端域名
  allowedHeaders: ['Authorization'],
  credentials: true
}));

// SSE 接口实现
app.get('/sse', (req, res) => {
  // 配置 SSE 响应头（必须）
  res.setHeader('Content-Type', 'text/event-stream');
  res.setHeader('Cache-Control', 'no-cache');
  res.setHeader('Connection', 'keep-alive');
  res.setHeader('X-Accel-Buffering', 'no'); // 禁用 Nginx 缓冲

  // 定时推送数据（模拟流式传输）
  let count = 0;
  const interval = setInterval(() => {
    count++;
    // 推送默认事件数据（message 事件）
    res.write(`data: ${JSON.stringify({ type: 'default', content: `消息${count}` })}\n\n`);

    // 每 5 次推送自定义事件数据
    if (count % 5 === 0) {
      res.write(`event: notification\ndata: ${JSON.stringify({ title: '通知', content: `第${count}条消息` })}\n\n`);
    }

    // 推送 10 次后停止
    if (count === 10) {
      clearInterval(interval);
      res.write(`event: complete\ndata: {"status": "done"}\n\n`);
      res.end();
    }
  }, 1000);

  // 客户端断开连接时清理资源
  req.on('close', () => {
    clearInterval(interval);
    res.end();
    console.log('客户端断开连接');
  });
});

// 启动服务器
app.listen(3001, () => {
  console.log('SSE 服务启动：http://localhost:3001');
});

3. 关键格式规范

服务器推送数据必须遵循 SSE 协议格式，否则客户端无法解析：

• 基础格式：data: 数据内容\n\n（末尾必须两个换行）
• 自定义事件格式：event: 事件名\ndata: 数据内容\n\n
• 多行数据：data: 第一行\ndata: 第二行\n\n（客户端会合并为一个字符串）
• 示例：

  plaintext

  // 正确格式
  data: {"name": "Alice"}\n\n
  
  // 自定义事件格式
  event: userLogin
  data: {"id": 1001, "name": "Alice"}\n\n
  
  // 多行数据格式
  data: 第一部分数据
  data: 第二部分数据
  data: 第三部分数据\n\n
  

五、框架集成实战

1. React 项目集成（AI 流式对话场景）

（1）组件实现

jsx

import React, { useState, useRef, useEffect } from 'react';
import EventSource from 'eventsource-polyfill';

const AIChatStream = () => {
  const [input, setInput] = useState('');
  const [chatHistory, setChatHistory] = useState([]);
  const [currentResponse, setCurrentResponse] = useState('');
  const esRef = useRef(null); // 存储 SSE 实例（避免重渲染丢失）
  const isStreamingRef = useRef(false); // 标记是否正在流式传输

  // 关闭 SSE 连接
  const closeSSE = () => {
    if (esRef.current) {
      esRef.current.close();
      esRef.current = null;
    }
    isStreamingRef.current = false;
  };

  // 发送 AI 请求并监听流式响应
  const sendRequest = () => {
    if (!input.trim() || isStreamingRef.current) return;

    // 更新对话历史
    setChatHistory(prev => [...prev, { role: 'user', content: input }]);
    setCurrentResponse('');
    setInput('');
    isStreamingRef.current = true;

    // 初始化 SSE 连接
    esRef.current = new EventSourcePolyfill(
      `http://localhost:3001/api/ai/stream?question=${encodeURIComponent(input)}`,
      {
        headers: { 'Authorization': 'Bearer your-ai-token' },
        heartbeatTimeout: 60000
      }
    );

    // 接收流式数据
    esRef.current.addEventListener('message', (e) => {
      const data = JSON.parse(e.data);
      if (data.content) {
        setCurrentResponse(prev => prev + data.content); // 逐字拼接响应
      }
    });

    // 生成完成事件
    esRef.current.addEventListener('complete', () => {
      setChatHistory(prev => [...prev, { role: 'ai', content: currentResponse }]);
      closeSSE();
    });

    // 错误处理
    esRef.current.addEventListener('error', (err) => {
      console.error('AI 流式错误：', err);
      setCurrentResponse('响应失败，请重试～');
      closeSSE();
    });
  };

  // 组件卸载时关闭连接
  useEffect(() => {
    return () => closeSSE();
  }, []);

  return (
    <div style={{ maxWidth: 800, margin: '0 auto', padding: 20 }}>
      <h3>AI 流式对话</h3>
      {/* 对话历史区域 */}
      <div style={{ 
        border: '1px solid #eee', 
        height: 400, 
        overflowY: 'auto', 
        padding: 10, 
        marginBottom: 10 
      }}>
        {chatHistory.map((item, idx) => (
          <div key={idx} style={{ margin: '10px 0', padding: 8 }}>
            <strong>{item.role === 'user' ? '你' : 'AI'}：</strong>
            <span>{item.content}</span>
          </div>
        ))}
        {/* 正在生成的 AI 响应 */}
        {isStreamingRef.current && (
          <div style={{ margin: '10px 0', padding: 8, backgroundColor: '#f5f5f5' }}>
            <strong>AI：</strong>
            <span>{currentResponse}</span>
            <span style={{ animation: 'blink 1s infinite' }}>▌</span>
          </div>
        )}
      </div>
      {/* 输入区域 */}
      <div style={{ display: 'flex' }}>
        <input
          type="text"
          value={input}
          onChange={(e) => setInput(e.target.value)}
          placeholder="输入你的问题..."
          style={{ flex: 1, padding: 8 }}
          disabled={isStreamingRef.current}
        />
        <button 
          onClick={sendRequest}
          style={{ marginLeft: 10, padding: '8px 16px' }}
          disabled={isStreamingRef.current || !input.trim()}
        >
          发送
        </button>
      </div>
      <style jsx global>{`
        @keyframes blink { 0%,100%{opacity:0} 50%{opacity:1} }
      `}</style>
    </div>
  );
};

export default AIChatStream;

（2）服务端 AI 接口实现（对接 OpenAI）

javascript

运行

const express = require('express');
const cors = require('cors');
const { OpenAI } = require('openai');
const app = express();

// 初始化 OpenAI 客户端
const openai = new OpenAI({ apiKey: 'your-openai-api-key' });

// 跨域配置
app.use(cors({
  origin: 'http://localhost:3000',
  allowedHeaders: ['Authorization'],
  credentials: true
}));

// AI 流式接口
app.get('/api/ai/stream', async (req, res) => {
  const { question } = req.query;
  const token = req.headers.authorization?.split(' ')[1];

  // 认证验证
  if (!token || token !== 'your-ai-token') {
    res.status(401).send('未授权');
    return;
  }

  // SSE 响应头配置
  res.setHeader('Content-Type', 'text/event-stream');
  res.setHeader('Cache-Control', 'no-cache');
  res.setHeader('Connection', 'keep-alive');
  res.setHeader('X-Accel-Buffering', 'no');

  try {
    // 调用 OpenAI 流式接口
    const stream = await openai.chat.completions.create({
      model: 'gpt-3.5-turbo',
      messages: [{ role: 'user', content: question }],
      stream: true,
      temperature: 0.7
    });

    // 流式推送 AI 响应
    for await (const chunk of stream) {
      const content = chunk.choices[0]?.delta?.content || '';
      if (content) {
        res.write(`data: ${JSON.stringify({ content })}\n\n`);
      }
    }

    // 生成完成
    res.write(`event: complete\ndata: {"status": "done"}\n\n`);
    res.end();
  } catch (err) {
    console.error('OpenAI 接口错误：', err);
    res.write(`event: error\ndata: {"message": "AI 生成失败"}\n\n`);
    res.end();
  }

  // 客户端断开连接时清理
  req.on('close', () => {
    stream?.destroy();
    res.end();
  });
});

app.listen(3001, () => {
  console.log('AI 流式服务启动：http://localhost:3001');
});

2. Vue 项目集成（实时日志监控场景）

vue

<template>
  <div class="log-monitor">
    <h3>实时日志监控</h3>
    <div class="log-container" ref="logContainer"></div>
    <button @click="startMonitor" :disabled="isMonitoring">开始监控</button>
    <button @click="stopMonitor" :disabled="!isMonitoring">停止监控</button>
  </div>
</template>

<script>
import EventSource from 'eventsource-polyfill';

export default {
  data() {
    return {
      es: null,
      isMonitoring: false
    };
  },
  methods: {
    startMonitor() {
      if (this.isMonitoring) return;

      this.isMonitoring = true;
      const logContainer = this.$refs.logContainer;

      // 初始化 SSE 连接
      this.es = new EventSourcePolyfill('http://localhost:3001/log/stream', {
        headers: { 'Authorization': 'Bearer your-token' }
      });

      // 接收日志数据
      this.es.addEventListener('message', (e) => {
        const log = JSON.parse(e.data);
        const logItem = document.createElement('div');
        logItem.className = `log-item log-${log.level}`;
        logItem.innerHTML = `[${new Date(log.timestamp).toLocaleString()}] [${log.level}] ${log.message}`;
        logContainer.appendChild(logItem);
        // 自动滚动到底部
        logContainer.scrollTop = logContainer.scrollHeight;
      });

      this.es.addEventListener('error', (err) => {
        console.error('日志监控错误：', err);
        this.stopMonitor();
      });
    },
    stopMonitor() {
      if (this.es) {
        this.es.close();
        this.es = null;
      }
      this.isMonitoring = false;
    }
  },
  beforeUnmount() {
    this.stopMonitor();
  }
};
</script>

<style scoped>
.log-monitor {
  max-width: 1200px;
  margin: 20px auto;
  padding: 20px;
}
.log-container {
  height: 500px;
  border: 1px solid #eee;
  padding: 10px;
  overflow-y: auto;
  margin: 10px 0;
  font-family: monospace;
}
.log-item {
  margin: 4px 0;
  padding: 4px;
}
.log-info { color: #333; }
.log-warn { color: #ff9800; }
.log-error { color: #f44336; }
button {
  margin-right: 10px;
  padding: 8px 16px;
  cursor: pointer;
}
button:disabled {
  cursor: not-allowed;
  opacity: 0.6;
}
</style>

六、高级优化技巧

1. 重连策略优化

默认重连机制可能无法满足复杂场景，可自定义重连逻辑：

javascript

运行

const createSSE = (url, options) => {
  let retryCount = 0;
  let es = new EventSourcePolyfill(url, options);

  es.addEventListener('error', (err) => {
    if (err.readyState === EventSource.CLOSED) {
      retryCount++;
      // 指数退避重连：1s → 2s → 4s → 8s（最多重试 5 次）
      const delay = Math.min(1000 * Math.pow(2, retryCount), 8000);
      setTimeout(() => {
        if (retryCount <= 5) {
          console.log(`第 ${retryCount} 次重连...`);
          es = createSSE(url, options); // 递归重建连接
        } else {
          console.log('重连失败，请刷新页面');
        }
      }, delay);
    }
  });

  return es;
};

// 使用自定义重连策略
const es = createSSE('http://localhost:3001/sse', {
  headers: { 'Authorization': 'Bearer your-token' }
});

2. 心跳检测优化

避免长连接因无数据传输被断开：

javascript

运行

// 前端配置
const es = new EventSourcePolyfill('http://localhost:3001/sse', {
  heartbeatTimeout: 60000 // 60秒超时
});

// 后端添加心跳推送
app.get('/sse', (req, res) => {
  // ... 基础响应头配置

  // 心跳定时器（每 20 秒推送一次空数据）
  const heartbeatInterval = setInterval(() => {
    res.write(`data: {"type": "heartbeat"}\n\n`);
  }, 20000);

  // 业务数据推送...

  // 连接关闭时清理
  req.on('close', () => {
    clearInterval(heartbeatInterval);
    res.end();
  });
});

3. 数据节流与防抖

应对高频数据推送导致的 UI 卡顿：

javascript

运行

import { debounce } from 'lodash';

// 防抖处理：50ms 内多次数据推送只更新一次 UI
const updateUI = debounce((data) => {
  console.log('更新 UI：', data);
  // 渲染逻辑...
}, 50);

// 监听数据时使用防抖函数
es.addEventListener('message', (e) => {
  const data = JSON.parse(e.data);
  updateUI(data);
});

4. 中断流式传输

支持用户主动取消请求：

javascript

运行

// React 组件中
const cancelStream = () => {
  if (esRef.current) {
    esRef.current.close();
    esRef.current = null;
    isStreamingRef.current = false;
    setCurrentResponse('已取消响应');
  }
};

// 服务端配合
app.get('/api/ai/stream', async (req, res) => {
  // ... 初始化逻辑

  // 监听客户端关闭事件
  req.on('close', () => {
    // 中断 AI 模型调用（以 OpenAI 为例）
    stream?.destroy();
    console.log('用户取消了请求');
    res.end();
  });
});

5. 跨域与安全优化

（1）严格 CORS 配置

javascript

运行

app.use(cors({
  origin: ['http://localhost:3000', 'https://your-domain.com'], // 白名单域名
  methods: ['GET'], // SSE 仅支持 GET 方法
  allowedHeaders: ['Authorization', 'Content-Type'],
  credentials: true,
  maxAge: 86400 // 预检请求缓存 24 小时
}));

（2）Token 认证优化

• 避免在 URL 中携带 Token（易被日志记录），通过 headers.Authorization 传递
• 实现 Token 过期自动刷新机制：

javascript

运行

es.addEventListener('error', (err) => {
  if (err.status === 401) {
    // Token 过期，尝试刷新 Token
    refreshToken().then(newToken => {
      // 重新建立连接
      es.close();
      createSSE('http://localhost:3001/sse', {
        headers: { 'Authorization': `Bearer ${newToken}` }
      });
    });
  }
});

七、常见问题排查

1. 跨域错误（CORS Error）

• 检查服务端是否配置 Access-Control-Allow-Origin 头
• 确保前端 withCredentials 与后端 credentials 配置一致
• 确认允许的请求头包含前端传递的自定义头（如 Authorization）

2. 数据无法解析（SyntaxError）

• 检查服务器推送格式是否符合 SSE 规范（末尾必须两个换行）
• 确保推送的 data 字段是合法的 JSON 字符串（避免未转义的特殊字符）
• 多行数据需每行添加 data: 前缀

3. 连接立即断开（readyState: CLOSED）

• 验证服务端响应头是否正确（Content-Type: text/event-stream）
• 检查服务端是否有数据推送（无数据推送会导致连接超时）
• 排查认证 Token 是否有效（401 错误会导致连接关闭）

4. 数据合并推送（失去流式效果）

• 后端添加 X-Accel-Buffering: no 头（禁用 Nginx 缓冲）
• 避免使用压缩中间件（如 compression），会缓冲响应
• 确保服务端每次 res.write 后立即刷新（Node.js 无需额外操作）

5. IE 浏览器不支持

• 确认已正确引入 EventSourcePolyfill（IE 无原生支持）
• IE 不支持 Promise，需额外引入 es6-promise polyfill：

bash

运行

npm install es6-promise --save

javascript

运行

import 'es6-promise/auto';
import EventSource from 'eventsource-polyfill';

八、总结与扩展

1. 核心要点总结

• EventSourcePolyfill 是跨浏览器 SSE 解决方案，兼容原生 API
• 流式数据传输的核心是服务器持续推送片段数据，客户端实时渲染
• 关键配置：响应头格式、跨域设置、心跳超时、认证 Token
• 优化重点：重连策略、数据节流、安全认证、中断机制

2. 与其他实时通信技术对比

技术	优势	劣势	适用场景
SSE（本文）	基于 HTTP、实现简单、跨域友好、轻量	单向通信、仅文本传输	实时推送、日志监控、AI 对话
WebSocket	双向通信、支持二进制、低延迟	实现复杂、跨域配置繁琐	即时聊天、协作工具
Polling（轮询）	兼容性极高、实现最简单	开销大、延迟高	简单实时需求、旧浏览器兼容

3. 扩展学习资源

• 官方文档：https://github.com/Yaffle/EventSource
• SSE 协议规范：https://html.spec.whatwg.org/multipage/server-sent-events.html
• OpenAI 流式 API：https://platform.openai.com/docs/api-reference/chat/create#chat-create-stream
• Nginx SSE 配置：https://nginx.org/en/docs/http/ngx_http_proxy_module.html#proxy_buffering