Server-Sent Events 流式推送

不是所有实时场景都需要 WebSocket——当数据只需要服务端 → 客户端单向推送时，SSE（Server-Sent Events）是更轻量的选择。它基于 HTTP、自动重连、浏览器原生支持。AI 视频生成进度、通知推送、日志流等场景非常适合。本章讲解 Nitro 中 SSE 的实现方式、与 WebSocket 的选型对比，以及断点续传机制。

1. SSE 基础与 createEventStream

1.1 SSE 协议要点

SSE 基于 HTTP 长连接，响应格式为纯文本流：

Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive

data: {"progress": 10, "stage": "初始化"}

data: {"progress": 45, "stage": "生成中"}

event: complete
data: {"progress": 100, "url": "/videos/abc.mp4"}

每条消息以 data: 开头，消息之间用空行分隔。可选的 event: 字段用于区分消息类型，id: 字段用于断点续传。

1.2 Nitro 中的 SSE 实现

Nitro 通过 H3 的 createEventStream 创建 SSE 流：

// server/api/sse/demo.get.ts
export default defineEventHandler(async (event) => {
  const stream = createEventStream(event)
 
  // 推送消息
  let count = 0
  const interval = setInterval(async () => {
    count++
    await stream.push({ data: JSON.stringify({ count, time: Date.now() }) })
 
    if (count >= 10) {
      clearInterval(interval)
      await stream.close()
    }
  }, 1000)
 
  // 客户端断开时清理
  stream.onClosed(() => {
    clearInterval(interval)
  })
 
  return stream.send()
})

1.3 客户端接收

浏览器原生 EventSource API：

const es = new EventSource('/api/sse/demo')
 
es.onmessage = (event) => {
  const data = JSON.parse(event.data)
  console.log(data)  // { count: 1, time: ... }
}
 
es.onerror = () => {
  console.log('连接断开，浏览器将自动重连')
}
 
// 关闭连接
es.close()

2. AI 视频生成进度推送

这是 SSE 最典型的应用场景——AI 视频生成可能需要数分钟，前端需要实时展示进度：

2.1 服务端

// server/api/videos/generate.post.ts
export default defineEventHandler(async (event) => {
  const user = await requireAuth(event)
  const { prompt, style, duration } = await readBody(event)
 
  // 创建任务记录
  const taskId = crypto.randomUUID()
  await useStorage('redis').setItem(`task:${taskId}`, {
    status: 'pending', userId: user.id, prompt,
  })
 
  // 触发异步生成（实际调用 AI API）
  generateVideoAsync(taskId, prompt, style, duration)
 
  return { taskId }
})
 
// server/api/videos/generate/[taskId]/progress.get.ts
export default defineEventHandler(async (event) => {
  const taskId = getRouterParam(event, 'taskId')!
  const stream = createEventStream(event)
 
  const interval = setInterval(async () => {
    const task = await useStorage('redis').getItem(`task:${taskId}`)
    if (!task) {
      await stream.push({ event: 'error', data: JSON.stringify({ message: '任务不存在' }) })
      await stream.close()
      return
    }
 
    await stream.push({
      id: String(Date.now()),  // 用于断点续传
      data: JSON.stringify({
        status: task.status,
        progress: task.progress || 0,
        stage: task.stage || '等待中',
        estimatedTime: task.estimatedTime,
      }),
    })
 
    if (task.status === 'completed' || task.status === 'failed') {
      clearInterval(interval)
      await stream.close()
    }
  }, 2000)
 
  stream.onClosed(() => clearInterval(interval))
  return stream.send()
})

2.2 客户端 Composable

// app/composables/useGenerateProgress.ts
export function useGenerateProgress(taskId: Ref<string | null>) {
  const progress = ref(0)
  const stage = ref('等待中')
  const status = ref<'pending' | 'processing' | 'completed' | 'failed'>('pending')
 
  let es: EventSource | null = null
 
  watch(taskId, (id) => {
    es?.close()
    if (!id) return
 
    es = new EventSource(`/api/videos/generate/${id}/progress`)
 
    es.onmessage = (event) => {
      const data = JSON.parse(event.data)
      progress.value = data.progress
      stage.value = data.stage
      status.value = data.status
 
      if (data.status === 'completed' || data.status === 'failed') {
        es?.close()
      }
    }
 
    es.onerror = () => { status.value = 'failed' }
  })
 
  onUnmounted(() => es?.close())
 
  return { progress, stage, status }
}

3. SSE vs WebSocket 选型

维度SSEWebSocket
方向服务端 → 客户端（单向）双向
协议HTTP独立协议（ws://）
自动重连✅ 浏览器内置❌ 需自己实现
数据格式文本（UTF-8）文本 + 二进制
并发连接HTTP/2 下无限制浏览器 ~256 个
代理兼容✅ 走标准 HTTP⚠️ 需 nginx 额外配置
Edge Runtime✅ 大部分支持⚠️ 部分不支持

选型建议：

  • 用 SSE：进度推送、通知流、日志流、AI 流式回复
  • 用 WebSocket：弹幕、聊天、协作编辑、游戏等双向通信

4. 断点续传与错误恢复

4.1 Last-Event-ID 机制

SSE 原生支持断点续传——浏览器重连时会自动发送 Last-Event-ID 请求头：

// 服务端：每条消息带 id
await stream.push({
  id: String(messageIndex),  // 消息序号
  data: JSON.stringify(payload),
})
 
// 服务端：重连时读取 Last-Event-ID
export default defineEventHandler(async (event) => {
  const lastId = getRequestHeader(event, 'last-event-id')
  const startFrom = lastId ? Number(lastId) + 1 : 0
 
  const stream = createEventStream(event)
 
  // 从断点处继续推送
  for (let i = startFrom; i < messages.length; i++) {
    await stream.push({ id: String(i), data: JSON.stringify(messages[i]) })
  }
 
  return stream.send()
})

4.2 自定义重连间隔

服务端可以通过 retry 字段控制客户端的重连间隔：

// 告诉客户端 5 秒后重连
await stream.push({ retry: 5000, data: '服务暂时繁忙' })

4.3 nginx 配置

SSE 需要禁用 nginx 的缓冲，否则数据会被攒到一起才发送：

location /api/sse/ {
    proxy_pass http://localhost:3000;
    proxy_set_header Connection '';
    proxy_http_version 1.1;
    proxy_buffering off;        # 关键：禁用缓冲
    proxy_cache off;
    chunked_transfer_encoding off;
}

5. AI 流式回复（ChatGPT 风格）

除了进度推送，SSE 的另一个核心场景是AI 对话流式输出——类似 ChatGPT 逐字输出的效果：

5.1 服务端：代理 OpenAI 流式 API

// server/api/ai/chat.post.ts
export default defineEventHandler(async (event) => {
  const user = await requireAuth(event)
  const { messages } = await readBody(event)
  const config = useRuntimeConfig()
 
  const stream = createEventStream(event)
 
  // 调用 OpenAI API（流式模式）
  const response = await fetch('https://api.openai.com/v1/chat/completions', {
    method: 'POST',
    headers: {
      'Authorization': `Bearer ${config.openaiApiKey}`,
      'Content-Type': 'application/json',
    },
    body: JSON.stringify({
      model: 'gpt-4',
      messages,
      stream: true,  // 关键：启用流式输出
    }),
  })
 
  if (!response.ok || !response.body) {
    await stream.push({ event: 'error', data: JSON.stringify({ message: 'AI 服务不可用' }) })
    await stream.close()
    return stream.send()
  }
 
  // 逐块读取 OpenAI 的 SSE 响应，转发给客户端
  const reader = response.body.getReader()
  const decoder = new TextDecoder()
  let buffer = ''
 
  try {
    while (true) {
      const { done, value } = await reader.read()
      if (done) break
 
      buffer += decoder.decode(value, { stream: true })
      const lines = buffer.split('\n')
      buffer = lines.pop() || ''  // 最后一行可能不完整
 
      for (const line of lines) {
        if (!line.startsWith('data: ')) continue
        const data = line.slice(6)
        if (data === '[DONE]') {
          await stream.push({ event: 'done', data: '' })
          break
        }
 
        const json = JSON.parse(data)
        const content = json.choices?.[0]?.delta?.content
        if (content) {
          await stream.push({ data: JSON.stringify({ content }) })
        }
      }
    }
  } finally {
    reader.releaseLock()
    await stream.close()
  }
 
  return stream.send()
})

5.2 客户端：逐字渲染

// app/composables/useAiChat.ts
export function useAiChat() {
  const answer = ref('')
  const isStreaming = ref(false)
 
  async function ask(messages: Array<{ role: string; content: string }>) {
    answer.value = ''
    isStreaming.value = true
 
    const es = new EventSource('/api/ai/chat?' + new URLSearchParams({
      _body: JSON.stringify({ messages }),  // SSE 不支持 POST body
    }))
    // 更好的做法：使用 fetch + ReadableStream（见 5.3）
 
    es.onmessage = (event) => {
      const { content } = JSON.parse(event.data)
      answer.value += content  // 逐字追加
    }
 
    es.addEventListener('done', () => {
      isStreaming.value = false
      es.close()
    })
 
    es.addEventListener('error', () => {
      isStreaming.value = false
      es.close()
    })
  }
 
  return { answer, isStreaming, ask }
}

5.3 fetch + ReadableStream（支持 POST）

EventSource 只支持 GET 请求，无法发送 POST body。对于 AI 对话这种需要发送消息列表的场景，推荐使用 fetch + ReadableStream：

// app/composables/useAiChat.ts — 改进版
export function useAiChat() {
  const answer = ref('')
  const isStreaming = ref(false)
 
  async function ask(messages: Array<{ role: string; content: string }>) {
    answer.value = ''
    isStreaming.value = true
 
    try {
      const response = await fetch('/api/ai/chat', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ messages }),
      })
 
      if (!response.ok || !response.body) throw new Error('请求失败')
 
      const reader = response.body.getReader()
      const decoder = new TextDecoder()
 
      while (true) {
        const { done, value } = await reader.read()
        if (done) break
 
        const text = decoder.decode(value, { stream: true })
        // 解析 SSE 格式
        for (const line of text.split('\n')) {
          if (!line.startsWith('data: ')) continue
          const data = line.slice(6)
          if (data === '[DONE]' || !data) continue
          const { content } = JSON.parse(data)
          if (content) answer.value += content
        }
      }
    } finally {
      isStreaming.value = false
    }
  }
 
  return { answer, isStreaming, ask }
}

6. 多事件类型与频道

6.1 自定义事件类型

SSE 支持 event 字段区分不同类型的消息：

// 服务端：推送不同类型的事件
await stream.push({ event: 'progress', data: JSON.stringify({ percent: 30 }) })
await stream.push({ event: 'log', data: '正在处理第 3 帧...' })
await stream.push({ event: 'preview', data: JSON.stringify({ url: '/preview/frame-3.jpg' }) })
await stream.push({ event: 'complete', data: JSON.stringify({ videoUrl: '/output.mp4' }) })
// 客户端：监听不同事件
const es = new EventSource('/api/videos/generate/xxx/progress')
 
es.addEventListener('progress', (e) => {
  progress.value = JSON.parse(e.data).percent
})
 
es.addEventListener('log', (e) => {
  logs.value.push(e.data)
})
 
es.addEventListener('preview', (e) => {
  previewUrl.value = JSON.parse(e.data).url
})
 
es.addEventListener('complete', (e) => {
  const { videoUrl } = JSON.parse(e.data)
  navigateTo(`/videos/${videoUrl}`)
  es.close()
})

6.2 通知流：多频道合并

一个 SSE 连接可以推送多种通知类型：

// server/api/notifications/stream.get.ts
export default defineEventHandler(async (event) => {
  const user = await requireAuth(event)
  const stream = createEventStream(event)
 
  const interval = setInterval(async () => {
    // 轮询用户的未读通知
    const notifications = await db.query.notifications.findMany({
      where: and(
        eq(notifications.userId, user.id),
        eq(notifications.isRead, false),
      ),
      orderBy: desc(notifications.createdAt),
      limit: 10,
    })
 
    for (const n of notifications) {
      await stream.push({
        event: n.type,  // 'comment', 'like', 'follow', 'system'
        id: n.id,
        data: JSON.stringify({
          title: n.title,
          body: n.body,
          link: n.link,
          createdAt: n.createdAt,
        }),
      })
    }
  }, 5000)
 
  stream.onClosed(() => clearInterval(interval))
  return stream.send()
})

7. 认证与 EventSource 限制

7.1 问题：EventSource 不支持自定义 Header

EventSource API 无法设置 Authorization 头，这是它最大的限制：

// ❌ EventSource 不支持
const es = new EventSource('/api/sse/protected', {
  headers: { 'Authorization': 'Bearer xxx' },  // 无此选项
})

7.2 解决方案

方案一：Cookie 认证（推荐）

Token 存在 httpOnly Cookie 中，SSE 请求会自动携带：

// 登录时设置 Cookie
setCookie(event, 'token', jwtToken, { httpOnly: true, secure: true, sameSite: 'strict' })
 
// SSE 接口从 Cookie 读取 Token（自动携带）
const token = parseCookies(event).token

方案二：URL 参数

const token = useCookie('auth-token').value
const es = new EventSource(`/api/sse/protected?token=${token}`)

方案三：用 fetch 替代 EventSource（见 5.3 节），可以自由设置请求头。

8. 背压与连接管理

8.1 背压（Backpressure）

如果服务端推送速度快于客户端消费速度（如网络慢），消息会堆积在缓冲区。处理策略：

// 服务端：检测缓冲区状态
const stream = createEventStream(event)
 
let messageQueue: string[] = []
const MAX_QUEUE_SIZE = 100
 
const interval = setInterval(async () => {
  const data = await fetchLatestData()
  messageQueue.push(JSON.stringify(data))
 
  // 如果队列过长，丢弃旧消息（保留最新的）
  if (messageQueue.length > MAX_QUEUE_SIZE) {
    const dropped = messageQueue.length - MAX_QUEUE_SIZE
    messageQueue = messageQueue.slice(dropped)
    console.warn(`Dropped ${dropped} messages due to backpressure`)
  }
 
  // 逐条发送（每条消息之间留一点间隔）
  while (messageQueue.length > 0) {
    const msg = messageQueue.shift()!
    try {
      await stream.push({ data: msg })
    } catch {
      // 发送失败说明连接已断开
      clearInterval(interval)
      return
    }
  }
}, 1000)

8.2 连接数限制

HTTP/1.1 浏览器对同一域名限制 6 个并发连接，SSE 会长期占用一个。在 HTTP/2 下无此限制（连接多路复用）。

最佳实践：

  • 确保服务启用 HTTP/2
  • 一个页面只建立一个 SSE 连接，通过 event 字段区分消息类型
  • 页面离开时及时 es.close() 释放连接

8.3 SSE 连接的服务端资源管理

每个 SSE 连接会占用一个服务端 TCP 连接。需要监控并限制最大同时连接数：

// server/utils/sse-manager.ts
let activeConnections = 0
const MAX_SSE_CONNECTIONS = 1000
 
export function acquireSseSlot() {
  if (activeConnections >= MAX_SSE_CONNECTIONS) {
    throw createError({ statusCode: 503, message: '服务繁忙，请稍后再试' })
  }
  activeConnections++
  return () => { activeConnections-- }  // 返回释放函数
}
 
// 使用
export default defineEventHandler(async (event) => {
  const release = acquireSseSlot()
  const stream = createEventStream(event)
  stream.onClosed(release)
  // ...
  return stream.send()
})

本章小结

  • createEventStream：Nitro/H3 创建 SSE 流的核心 API，push 推送消息，close 关闭流
  • AI 生成进度：异步任务 + Redis 状态存储 + SSE 轮询推送
  • AI 流式回复：代理 OpenAI 流式 API，逐块转发给客户端，实现 ChatGPT 风格的逐字输出
  • fetch vs EventSource：EventSource 只支持 GET，AI 对话场景用 fetch + ReadableStream
  • 多事件类型：event 字段区分消息类型，客户端 addEventListener 分别监听
  • 认证：Cookie 认证最简单，URL 参数次之，fetch 方式最灵活
  • 选型对比：单向推送用 SSE（更简单、自动重连），双向通信用 WebSocket
  • 断点续传：SSE 原生支持 id + Last-Event-ID，浏览器自动重连并续传
  • 背压处理：监控消息队列长度，超限时丢弃旧消息
  • 部署注意：nginx 需禁用 proxy_buffering，服务端需限制最大 SSE 连接数