WebSocket 实时通信

视频平台的弹幕、实时评论、协作编辑等场景需要服务端主动推送数据。Nitro 内置了 WebSocket 支持——通过 defineWebSocketHandler 定义处理器，无需引入 Socket.IO 等第三方库。本章讲解 Nitro WebSocket 的完整用法，并以视频弹幕系统为实战案例。

1. Nitro WebSocket 基础

1.1 启用 WebSocket

// nuxt.config.ts
export default defineNuxtConfig({
  nitro: {
    experimental: {
      websocket: true,  // 启用 WebSocket 支持
    },
  },
})

1.2 defineWebSocketHandler

WebSocket 路由放在 server/routes/ 目录下（不是 api/），使用 defineWebSocketHandler 定义：

// server/routes/ws/chat.ts → ws://localhost:3000/ws/chat
export default defineWebSocketHandler({
  open(peer) {
    // 连接建立时
    console.log(`[WebSocket] 新连接: ${peer.id}`)
  },
 
  message(peer, message) {
    // 收到消息时
    const text = message.text()
    console.log(`[WebSocket] ${peer.id}: ${text}`)
 
    // 回复消息
    peer.send(JSON.stringify({ type: 'echo', data: text }))
  },
 
  close(peer, details) {
    // 连接关闭时
    console.log(`[WebSocket] 断开: ${peer.id}, code: ${details.code}`)
  },
 
  error(peer, error) {
    // 错误处理
    console.error(`[WebSocket] 错误: ${peer.id}`, error)
  },
})

1.3 Peer 对象

peer 是每个 WebSocket 连接的抽象，核心 API：

方法/属性说明
peer.id连接唯一标识
peer.send(data)发送消息（字符串或 Buffer）
peer.close(code?, reason?)关闭连接
peer.subscribe(channel)订阅频道
peer.unsubscribe(channel)取消订阅
peer.publish(channel, data)向频道广播（不包含自己）

2. AI 视频弹幕实时推送

2.1 服务端：弹幕频道

// server/routes/ws/danmaku/[videoId].ts
export default defineWebSocketHandler({
  open(peer) {
    const videoId = peer.url?.split('/').pop()
    if (!videoId) return peer.close(1008, '缺少 videoId')
 
    // 订阅该视频的弹幕频道
    peer.subscribe(`danmaku:${videoId}`)
    peer.send(JSON.stringify({ type: 'connected', videoId }))
  },
 
  async message(peer, message) {
    try {
      const { videoId, text, time, color } = JSON.parse(message.text())
 
      // 校验弹幕内容
      if (!text || text.length > 100) return
      if (typeof time !== 'number') return
 
      const danmaku = { id: Date.now(), text, time, color: color || '#ffffff', ts: Date.now() }
 
      // 广播给同一视频的所有观众
      peer.publish(`danmaku:${videoId}`, JSON.stringify({ type: 'danmaku', data: danmaku }))
 
      // 同时回发给发送者自己
      peer.send(JSON.stringify({ type: 'danmaku', data: danmaku }))
 
      // 持久化到数据库（异步，不阻塞响应）
      db.insert(danmakus).values({ videoId, ...danmaku }).catch(console.error)
    } catch {}
  },
 
  close(peer) {
    // peer 关闭时自动取消所有订阅，无需手动处理
  },
})

2.2 客户端：连接与发送

// app/composables/useDanmaku.ts
export function useDanmaku(videoId: string) {
  const danmakuList = ref<Danmaku[]>([])
  const isConnected = ref(false)
  let ws: WebSocket | null = null
 
  function connect() {
    if (import.meta.server) return
    const protocol = location.protocol === 'https:' ? 'wss:' : 'ws:'
    ws = new WebSocket(`${protocol}//${location.host}/ws/danmaku/${videoId}`)
 
    ws.onopen = () => { isConnected.value = true }
    ws.onclose = () => {
      isConnected.value = false
      setTimeout(connect, 3000)  // 自动重连
    }
    ws.onmessage = (e) => {
      const msg = JSON.parse(e.data)
      if (msg.type === 'danmaku') danmakuList.value.push(msg.data)
    }
  }
 
  function send(text: string, time: number, color?: string) {
    ws?.send(JSON.stringify({ videoId, text, time, color }))
  }
 
  onMounted(connect)
  onUnmounted(() => ws?.close())
 
  return { danmakuList, isConnected, send }
}

3. 连接管理与房间分组

3.1 Pub/Sub 频道模式

Nitro WebSocket 的 subscribe / publish 实现了发布-订阅模式，天然支持房间分组：

// 用户加入视频房间
peer.subscribe(`video:${videoId}`)           // 弹幕频道
peer.subscribe(`video:${videoId}:comments`)  // 评论频道
peer.subscribe(`user:${userId}`)             // 个人通知频道
 
// 广播消息到频道（不包含发送者自己）
peer.publish(`video:${videoId}`, JSON.stringify({ type: 'viewer_count', count: 42 }))

3.2 在线人数统计

// server/routes/ws/room/[videoId].ts
const roomCounts = new Map<string, number>()
 
export default defineWebSocketHandler({
  open(peer) {
    const videoId = peer.url?.split('/').pop()!
    peer.subscribe(`room:${videoId}`)
 
    const count = (roomCounts.get(videoId) || 0) + 1
    roomCounts.set(videoId, count)
 
    // 广播最新人数
    peer.publish(`room:${videoId}`, JSON.stringify({ type: 'count', count }))
    peer.send(JSON.stringify({ type: 'count', count }))
  },
 
  close(peer) {
    const videoId = peer.url?.split('/').pop()!
    const count = Math.max(0, (roomCounts.get(videoId) || 1) - 1)
    roomCounts.set(videoId, count)
    peer.publish(`room:${videoId}`, JSON.stringify({ type: 'count', count }))
  },
})

4. 断线重连与心跳检测

4.1 客户端重连策略

推荐指数退避重连，避免服务器重启时被大量连接冲击：

function createReconnect(connectFn: () => void) {
  let retries = 0
  const maxRetries = 10
  const baseDelay = 1000  // 1 秒
 
  return function reconnect() {
    if (retries >= maxRetries) {
      console.error('WebSocket 重连次数已达上限')
      return
    }
    const delay = Math.min(baseDelay * 2 ** retries, 30000) // 最大 30 秒
    retries++
    setTimeout(connectFn, delay)
  }
}

4.2 心跳检测

长时间无数据传输时，中间代理（nginx、CDN）可能会断开连接。心跳保活：

// 客户端每 30 秒发送 ping
let heartbeatTimer: ReturnType<typeof setInterval>
 
ws.onopen = () => {
  heartbeatTimer = setInterval(() => {
    if (ws?.readyState === WebSocket.OPEN) {
      ws.send(JSON.stringify({ type: 'ping' }))
    }
  }, 30000)
}
 
ws.onclose = () => {
  clearInterval(heartbeatTimer)
}
 
// 服务端回复 pong
// server 端在 message handler 中：
if (data.type === 'ping') {
  peer.send(JSON.stringify({ type: 'pong' }))
  return
}

5. WebSocket 鉴权

WebSocket 握手是一个标准 HTTP 请求，可以在此阶段做身份验证。但 defineWebSocketHandler 的 open 回调拿到的已经是升级后的连接，无法直接 reject HTTP。推荐的做法有两种：

5.1 方案一：URL 参数传 Token

客户端在连接 URL 中附带 Token：

// 客户端
const token = useCookie('token').value
const ws = new WebSocket(`wss://example.com/ws/chat?token=${token}`)
// server/routes/ws/chat.ts
export default defineWebSocketHandler({
  open(peer) {
    // 从 URL 查询参数中提取 Token
    const url = new URL(peer.url || '', 'http://localhost')
    const token = url.searchParams.get('token')
 
    if (!token) {
      peer.send(JSON.stringify({ type: 'error', message: '未提供认证凭证' }))
      peer.close(4001, 'Unauthorized')
      return
    }
 
    try {
      const config = useRuntimeConfig()
      const payload = jwt.verify(token, config.jwtSecret) as { userId: string; role: string }
      // 将用户信息存入 peer 的自定义上下文
      peer.ctx = { user: payload }
      peer.subscribe(`user:${payload.userId}`)
    } catch {
      peer.send(JSON.stringify({ type: 'error', message: 'Token 无效' }))
      peer.close(4001, 'Invalid token')
    }
  },
 
  message(peer, message) {
    if (!peer.ctx?.user) return  // 未认证的连接忽略消息
    // ... 处理消息
  },
})

5.2 方案二：先 HTTP 认证再升级

用一个普通 API 接口验证身份、生成一次性 ticket，客户端再用 ticket 连接 WebSocket：

// 1. 客户端先请求 ticket
const { ticket } = await $fetch('/api/ws/ticket', { method: 'POST' })
 
// 2. 用 ticket 连接 WebSocket（ticket 只能用一次，5 分钟过期）
const ws = new WebSocket(`wss://example.com/ws/chat?ticket=${ticket}`)
// server/api/ws/ticket.post.ts
export default defineEventHandler(async (event) => {
  const user = await requireAuth(event)
  const ticket = crypto.randomUUID()
 
  // 存储 ticket → 用户映射（5 分钟过期）
  await useStorage('redis').setItem(`ws-ticket:${ticket}`, {
    userId: user.id,
    role: user.role,
  }, { ttl: 300 })
 
  return { ticket }
})

这种方案更安全——Token 不会出现在 URL 中（URL 可能被日志记录），且 ticket 是一次性的。

6. 消息协议设计

生产级 WebSocket 应用需要一套规范的消息协议，避免随意拼 JSON：

6.1 统一消息格式

// 共享类型（客户端和服务端都用）
interface WsMessage<T = unknown> {
  type: string         // 消息类型
  data?: T             // 消息数据
  id?: string          // 消息 ID（用于 ACK 确认）
  timestamp?: number   // 时间戳
  error?: string       // 错误信息
}
 
// 常见消息类型枚举
const WS_TYPES = {
  // 系统消息
  PING: 'ping',
  PONG: 'pong',
  ERROR: 'error',
  ACK: 'ack',
 
  // 弹幕
  DANMAKU_SEND: 'danmaku:send',
  DANMAKU_NEW: 'danmaku:new',
 
  // 通知
  NOTIFICATION: 'notification',
 
  // 房间
  ROOM_JOIN: 'room:join',
  ROOM_LEAVE: 'room:leave',
  ROOM_COUNT: 'room:count',
} as const

6.2 服务端消息路由

根据消息类型分发处理逻辑，避免一个巨大的 message 回调：

// server/utils/ws-router.ts
type MessageHandler = (peer: any, data: any) => void | Promise<void>
 
export function createWsRouter() {
  const handlers = new Map<string, MessageHandler>()
 
  return {
    on(type: string, handler: MessageHandler) {
      handlers.set(type, handler)
    },
 
    async dispatch(peer: any, rawMessage: any) {
      try {
        const msg: WsMessage = JSON.parse(rawMessage.text())
 
        // 心跳快速响应
        if (msg.type === 'ping') {
          peer.send(JSON.stringify({ type: 'pong' }))
          return
        }
 
        const handler = handlers.get(msg.type)
        if (!handler) {
          peer.send(JSON.stringify({ type: 'error', error: `未知消息类型: ${msg.type}` }))
          return
        }
 
        await handler(peer, msg.data)
 
        // 如果消息带 ID，回复 ACK
        if (msg.id) {
          peer.send(JSON.stringify({ type: 'ack', id: msg.id }))
        }
      } catch (err) {
        peer.send(JSON.stringify({ type: 'error', error: '消息格式错误' }))
      }
    },
  }
}
// server/routes/ws/video/[videoId].ts
const router = createWsRouter()
 
router.on('danmaku:send', async (peer, data) => {
  const { text, time, color } = data
  // 校验 + 广播 + 持久化...
})
 
router.on('room:join', (peer, data) => {
  peer.subscribe(`room:${data.roomId}`)
})
 
export default defineWebSocketHandler({
  open(peer) { /* 鉴权 + 初始化 */ },
  message(peer, message) { router.dispatch(peer, message) },
  close(peer) { /* 清理 */ },
})

7. 二进制数据传输

WebSocket 不仅能传文本，还能传二进制数据。适用于音频流、实时视频帧等场景：

// 服务端：发送二进制数据
export default defineWebSocketHandler({
  message(peer, message) {
    // 判断消息类型
    if (message.isBinary) {
      const buffer = message.rawData  // ArrayBuffer
      // 处理二进制数据（如音频帧）
      console.log(`收到二进制数据: ${buffer.byteLength} bytes`)
 
      // 原样广播给同频道其他 peer
      peer.publish('audio-stream', buffer)
    } else {
      // 文本消息走 JSON 协议
      const data = JSON.parse(message.text())
      // ...
    }
  },
})
// 客户端：发送二进制数据
ws.binaryType = 'arraybuffer'  // 设置接收格式
 
// 发送音频数据
const audioBuffer = new ArrayBuffer(1024)
ws.send(audioBuffer)
 
// 接收二进制数据
ws.onmessage = (event) => {
  if (event.data instanceof ArrayBuffer) {
    // 处理二进制数据
    const audioData = new Float32Array(event.data)
  } else {
    // 处理文本消息
    const msg = JSON.parse(event.data)
  }
}

8. 多实例部署与 Redis 扩展

Nitro 内置的 Pub/Sub 是进程内的——如果部署了多个 Nitro 实例（如 K8s 多 Pod），不同实例之间的 peer 无法互相通信。解决方案是引入 Redis Pub/Sub 作为消息总线：

┌──────────────┐  Redis Pub/Sub  ┌──────────────┐
│  Nitro 实例 A │◄──────────────►│  Nitro 实例 B │
│  peer 1, 2   │                 │  peer 3, 4   │
└──────────────┘                 └──────────────┘

// server/plugins/ws-redis-bridge.ts
import Redis from 'ioredis'
 
export default defineNitroPlugin((nitroApp) => {
  const config = useRuntimeConfig()
  const pub = new Redis(config.redisUrl)
  const sub = new Redis(config.redisUrl)
 
  // 全局消息桥接
  const localChannels = new Set<string>()
 
  // 当本地有 peer 发布消息时，同步到 Redis
  nitroApp.hooks.hook('websocket:message', (peer, message) => {
    // 只转发需要跨实例同步的频道消息
    const msg = JSON.parse(message.text())
    if (msg.broadcast && msg.channel) {
      pub.publish(`ws:${msg.channel}`, JSON.stringify({
        ...msg,
        fromInstance: process.env.INSTANCE_ID,
      }))
    }
  })
 
  // 从 Redis 接收其他实例的消息，广播给本地 peer
  sub.on('message', (channel, data) => {
    const msg = JSON.parse(data)
    // 跳过自己发出的消息
    if (msg.fromInstance === process.env.INSTANCE_ID) return
 
    const wsChannel = channel.replace('ws:', '')
    // 利用 Nitro 的 publish 广播给本地订阅了该频道的 peer
    nitroApp.hooks.callHook('websocket:publish', wsChannel, data)
  })
 
  // 优雅关闭
  nitroApp.hooks.hook('close', async () => {
    await pub.quit()
    await sub.quit()
  })
})

简化方案：如果不想自己实现 Redis 桥接，可以用 Socket.IO + @socket.io/redis-adapter，它内置了 Redis 多实例同步。但代价是引入了额外的协议层和包体积。

9. nginx WebSocket 代理配置

生产环境中 nginx 需要额外配置才能正确代理 WebSocket：

# nginx.conf
map $http_upgrade $connection_upgrade {
    default upgrade;
    ''      close;
}
 
server {
    listen 443 ssl;
    server_name example.com;
 
    location /ws/ {
        proxy_pass http://localhost:3000;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;          # 关键：协议升级
        proxy_set_header Connection $connection_upgrade;  # 关键：保持连接
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
 
        proxy_read_timeout 86400s;   # 24 小时超时（默认 60s 会断连）
        proxy_send_timeout 86400s;
    }
}

关键配置：

  • Upgrade 和 Connection 头是 WebSocket 握手的必要条件
  • proxy_read_timeout 必须设置足够大，否则空闲连接会被 nginx 主动断开

本章小结

  • defineWebSocketHandler：Nitro 内置的 WebSocket 支持，放在 server/routes/ 下
  • Peer 对象：send（发送）、subscribe/publish（发布-订阅频道）、close（关闭）
  • 弹幕系统：频道隔离（按视频 ID 订阅）、广播推送、异步持久化
  • 房间管理：Pub/Sub 模式天然支持房间分组和在线人数统计
  • 鉴权：URL 参数 Token 或一次性 ticket 机制，在 open 回调中验证
  • 消息协议：统一 &#123; type, data, id &#125; 格式 + 消息路由器，避免巨型 message 回调
  • 二进制数据：message.isBinary 判断，ws.binaryType = 'arraybuffer' 接收
  • 多实例扩展：Redis Pub/Sub 作为跨实例消息总线
  • 可靠性：客户端指数退避重连 + 心跳保活 + nginx 超时配置