实时通信：WebSocket 与 SSE

AI 对话的流式回复、多人协作的实时同步、后台任务的进度推送——这些场景都需要服务端主动向客户端推送数据。本章对比 WebSocket 和 SSE 两种方案，讲清在 Next.js 中的实现方式和选择策略。

1. 方案对比

维度SSEWebSocket
方向服务端 → 客户端（单向）双向
协议HTTP独立协议（ws://）
实现难度低（标准 Response）高（需要额外服务器）
Next.js 原生支持✅ Route Handler❌ 需要自定义服务器
自动重连✅ 浏览器内置❌ 需要手动实现
Edge Runtime✅❌
适用场景AI 流式回复、通知推送、进度更新聊天室、协同编辑、游戏

核心结论：Next.js App Router 中优先用 SSE。WebSocket 只在需要双向通信时使用，且需要额外的 WebSocket 服务器。

2. Server-Sent Events (SSE)

2.1 原理

SSE 基于 HTTP 长连接——客户端发起一个 GET 请求，服务端保持连接不关闭，持续推送数据。数据格式是纯文本，每条消息以 data: 开头，以两个换行符结尾：

data: {"type": "token", "content": "Hello"}

data: {"type": "token", "content": " World"}

data: {"type": "done"}

2.2 服务端实现

在 Route Handler 中用 ReadableStream 创建 SSE 响应：

// app/api/sse/notifications/route.ts
export async function GET(request: Request) {
  const encoder = new TextEncoder()
 
  const stream = new ReadableStream({
    async start(controller) {
      // 发送一条 SSE 消息的辅助函数
      function send(data: unknown) {
        controller.enqueue(encoder.encode(`data: ${JSON.stringify(data)}\n\n`))
      }
 
      // 发送初始连接确认
      send({ type: 'connected', timestamp: Date.now() })
 
      // 模拟：每 5 秒推送一条通知
      const interval = setInterval(() => {
        send({ type: 'notification', message: '新消息', timestamp: Date.now() })
      }, 5000)
 
      // 客户端断开时清理
      request.signal.addEventListener('abort', () => {
        clearInterval(interval)
        controller.close()
      })
    },
  })
 
  return new Response(stream, {
    headers: {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache, no-transform',
      Connection: 'keep-alive',
    },
  })
}

2.3 客户端消费

浏览器原生提供 EventSource API：

'use client'
 
import { useEffect, useState } from 'react'
 
export function NotificationListener() {
  const [notifications, setNotifications] = useState<string[]>([])
 
  useEffect(() => {
    const source = new EventSource('/api/sse/notifications')
 
    source.onmessage = (event) => {
      const data = JSON.parse(event.data)
      if (data.type === 'notification') {
        setNotifications((prev) => [...prev, data.message])
      }
    }
 
    source.onerror = () => {
      // EventSource 会自动重连，无需手动处理
      console.log('SSE 连接断开，自动重连中...')
    }
 
    return () => source.close()
  }, [])
 
  return (
    <ul>
      {notifications.map((msg, i) => <li key={i}>{msg}</li>)}
    </ul>
  )
}

EventSource 的优势：浏览器自动处理重连。连接断开后会自动重新发起请求，对网络不稳定的场景很友好。

2.4 带认证的 SSE

EventSource 不支持自定义 Header（无法传 Authorization）。解决方案：

方案一：URL 中传 Token

const source = new EventSource(`/api/sse/notifications?token=${sessionToken}`)

服务端从 URL 参数校验 Token。简单但不够安全（Token 会出现在日志中）。

方案二：用 fetch 替代 EventSource

async function connectSSE() {
  const response = await fetch('/api/sse/notifications', {
    headers: { Authorization: `Bearer ${token}` },
  })
 
  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)
    // 解析 SSE 格式：去掉 "data: " 前缀
    const lines = text.split('\n').filter((line) => line.startsWith('data: '))
    for (const line of lines) {
      const data = JSON.parse(line.slice(6))
      handleMessage(data)
    }
  }
}

方案三：Cookie 认证（推荐）

如果用 NextAuth / Lucia 等基于 Cookie 的认证，EventSource 会自动携带 Cookie——无需额外处理。服务端用 cookies() 或 auth() 校验即可。

3. AI 流式输出（SSE 实战）

AI 对话的流式输出是 SSE 最典型的应用。第 26 章已经展示了基础的流式 Route Handler，这里补充结构化 SSE 协议——不仅传文本内容，还传元数据（Token 计数、完成信号、错误信息）：

// app/api/chat/route.ts — 结构化 SSE
export async function POST(request: Request) {
  const { messages, model } = await request.json()
  const encoder = new TextEncoder()
 
  const stream = new ReadableStream({
    async start(controller) {
      function sendSSE(event: string, data: unknown) {
        controller.enqueue(encoder.encode(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`))
      }
 
      try {
        const completion = await openai.chat.completions.create({
          model, messages, stream: true, stream_options: { include_usage: true },
        })
 
        for await (const chunk of completion) {
          const content = chunk.choices[0]?.delta?.content
          if (content) {
            sendSSE('token', { content })
          }
 
          // 最后一个 chunk 包含 usage 信息
          if (chunk.usage) {
            sendSSE('usage', {
              inputTokens: chunk.usage.prompt_tokens,
              outputTokens: chunk.usage.completion_tokens,
            })
          }
        }
 
        sendSSE('done', { finishReason: 'stop' })
      } catch (error) {
        sendSSE('error', { message: (error as Error).message })
      } finally {
        controller.close()
      }
    },
  })
 
  return new Response(stream, {
    headers: { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache' },
  })
}

客户端用 EventSource 监听不同事件类型：

const source = new EventSource('/api/chat')
 
source.addEventListener('token', (e) => {
  const { content } = JSON.parse(e.data)
  appendToMessage(content)
})
 
source.addEventListener('usage', (e) => {
  const { inputTokens, outputTokens } = JSON.parse(e.data)
  updateTokenCount(inputTokens, outputTokens)
})
 
source.addEventListener('done', () => {
  source.close()
  setStatus('idle')
})
 
source.addEventListener('error', (e) => {
  const { message } = JSON.parse(e.data)
  setError(message)
  source.close()
})
AI SDK 的 Data Stream 协议

Vercel AI SDK 的 toDataStreamResponse() 本质上就是一个结构化的 SSE 流，内部定义了 0: (text)、2: (data)、d: (done) 等前缀协议。如果用 AI SDK，不需要自己实现这套协议。

4. WebSocket

4.1 Next.js 的限制

Next.js 的开发服务器和 Vercel 部署都不原生支持 WebSocket。WebSocket 需要持久的双向连接，而 Serverless 函数是请求-响应模式。

要用 WebSocket，有三种方案：

方案特点
自定义 Node.js 服务器完全控制，但失去 Vercel 部署优势
独立 WebSocket 服务Next.js + 单独的 WS 服务器（如 Socket.io 独立部署）
第三方服务Pusher、Ably、PartyKit、Liveblocks

4.2 Socket.io 独立服务

最常见的模式：Next.js 负责页面和 API，Socket.io 服务器独立部署处理实时通信：

// ws-server.ts — 独立的 Socket.io 服务器
import { Server } from 'socket.io'
import { createServer } from 'http'
 
const httpServer = createServer()
const io = new Server(httpServer, {
  cors: { origin: process.env.NEXT_PUBLIC_APP_URL },
})
 
io.on('connection', (socket) => {
  console.log('Client connected:', socket.id)
 
  // 加入工作空间房间
  socket.on('join-workspace', (workspaceId: string) => {
    socket.join(`workspace:${workspaceId}`)
  })
 
  // 广播消息到工作空间
  socket.on('chat-message', (data) => {
    socket.to(`workspace:${data.workspaceId}`).emit('new-message', data)
  })
 
  socket.on('disconnect', () => {
    console.log('Client disconnected:', socket.id)
  })
})
 
httpServer.listen(3001)

客户端连接：

'use client'
 
import { io, Socket } from 'socket.io-client'
import { useEffect, useRef } from 'react'
 
export function useSocket(workspaceId: string) {
  const socketRef = useRef<Socket | null>(null)
 
  useEffect(() => {
    const socket = io(process.env.NEXT_PUBLIC_WS_URL!, {
      transports: ['websocket'],
    })
 
    socket.on('connect', () => {
      socket.emit('join-workspace', workspaceId)
    })
 
    socket.on('new-message', (data) => {
      // 收到其他用户的消息 → 更新 UI
      useChatStore.getState().addMessage(data)
    })
 
    socketRef.current = socket
    return () => { socket.disconnect() }
  }, [workspaceId])
 
  return socketRef
}

4.3 PartyKit（推荐）

PartyKit 是专为 Serverless 设计的实时通信平台——和 Vercel 部署配合好，不需要自己管服务器：

// party/chat.ts — PartyKit 服务端
import type * as Party from 'partykit/server'
 
export default class ChatRoom implements Party.Server {
  constructor(readonly room: Party.Room) {}
 
  onMessage(message: string, sender: Party.Connection) {
    // 广播给房间内所有其他连接
    this.room.broadcast(message, [sender.id])
  }
 
  onConnect(conn: Party.Connection) {
    conn.send(JSON.stringify({ type: 'welcome', id: conn.id }))
  }
}
// 客户端连接
import usePartySocket from 'partysocket/react'
 
export function ChatRoom({ chatId }: { chatId: string }) {
  const socket = usePartySocket({
    host: process.env.NEXT_PUBLIC_PARTYKIT_HOST!,
    room: chatId,
    onMessage(event) {
      const data = JSON.parse(event.data)
      // 处理实时消息
    },
  })
 
  function sendMessage(content: string) {
    socket.send(JSON.stringify({ type: 'message', content }))
  }
 
  return <ChatInput onSend={sendMessage} />
}

5. 选择策略

场景推荐方案原因
AI 流式回复SSE（Route Handler）单向推送，Next.js 原生支持
后台任务进度SSE服务端推送进度百分比
通知推送SSE服务端单向通知
多人实时聊天WebSocket（PartyKit）需要双向通信
协同编辑WebSocket（Liveblocks）需要 CRDT + 双向同步
实时光标位置WebSocket（Liveblocks）高频双向更新

决策流程：

  1. 只需要服务端推送？→ SSE
  2. 需要双向通信？→ WebSocket
  3. 部署在 Vercel？→ PartyKit / Ably / Pusher
  4. 自托管？→ Socket.io

6. SSE vs Polling

有人会问：为什么不用轮询（每隔 N 秒请求一次）？

维度SSE轮询
延迟实时（毫秒级）取决于轮询间隔
资源消耗1 个长连接每次都建立新连接
服务端压力低高（大量无效请求）
实现难度稍高低

轮询唯一的优势是简单。如果数据更新频率低（每分钟一次）且对延迟不敏感，轮询也可以。但 AI 流式回复、实时通知这类场景必须用 SSE 或 WebSocket。

本章小结

  • SSE 是 Next.js 中实时推送的首选——Route Handler 原生支持，Edge 兼容
  • 结构化 SSE：用 event: + data: 区分消息类型（token/usage/done/error）
  • WebSocket 用于双向通信——Next.js 不原生支持，需要独立服务或第三方（PartyKit）
  • 认证：SSE 优先用 Cookie 认证，避免 URL 传 Token
  • AI 流式：简单场景用 ReadableStream，复杂场景用 AI SDK 的 toDataStreamResponse()

下一章讲后台任务与队列。