WebSocket 完全指南

官方文档：https://developer.mozilla.org/zh-CN/docs/Web/API/WebSockets_API
适用版本：WebSocket API（Living Standard，2026-05-07 核实）

WebSocket 是一种在单个 TCP 连接上提供全双工通信的协议，适用于需要服务端主动推送数据的场景（实时聊天、行情推送、协作编辑等）。

原生 WebSocket API

构造函数参数

// 基础连接
const ws = new WebSocket('ws://localhost:8000/ws')

// 指定子协议
const ws = new WebSocket('wss://api.example.com/ws', ['chat', 'v2'])

readyState 枚举值

事件

MessageEvent 字段：

CloseEvent 字段：

const ws = new WebSocket('ws://localhost:8000/ws')

ws.onopen = (event) => {
  console.log('连接已建立')
  ws.send(JSON.stringify({ type: 'hello' }))
}

ws.onmessage = (event) => {
  const data = JSON.parse(event.data as string)
  console.log('收到消息:', data)
}

ws.onerror = (event) => {
  console.error('WebSocket 错误:', event)
}

ws.onclose = (event) => {
  console.log(`连接关闭，code: ${event.code}, reason: ${event.reason}`)
}

send 支持的数据类型

// 文本消息
ws.send(JSON.stringify({ type: 'chat', content: 'Hello' }))

// 二进制消息
const buffer = new ArrayBuffer(4)
const view = new DataView(buffer)
view.setUint32(0, 12345)
ws.send(buffer)

close 关闭码

// 正常关闭
ws.close(1000, 'Normal closure')

// 应用层主动断开
ws.close(1001, 'Going away')

封装实践

完整封装（自动重连 + 心跳 + 消息队列）

interface WebSocketOptions {
  url: string
  // 最大重连次数，默认 5
  maxRetries?: number
  // 初始重连延迟（毫秒），指数退避基础值，默认 1000
  retryDelay?: number
  // 心跳间隔（毫秒），默认 30000，设为 0 禁用
  heartbeatInterval?: number
  // 心跳消息内容
  heartbeatMessage?: string
  // 消息接收回调
  onMessage?: (data: unknown) => void
  onOpen?: () => void
  onClose?: (event: CloseEvent) => void
  onError?: (event: Event) => void
}

class ReconnectingWebSocket {
  private ws: WebSocket | null = null
  private retryCount = 0
  private heartbeatTimer: ReturnType<typeof setInterval> | null = null
  private messageQueue: string[] = []
  private destroyed = false

  private readonly options: Required<WebSocketOptions>

  constructor(options: WebSocketOptions) {
    this.options = {
      maxRetries: 5,
      retryDelay: 1000,
      heartbeatInterval: 30000,
      heartbeatMessage: JSON.stringify({ type: 'ping' }),
      onMessage: () => {},
      onOpen: () => {},
      onClose: () => {},
      onError: () => {},
      ...options
    }
    this.connect()
  }

  private connect() {
    if (this.destroyed) return

    this.ws = new WebSocket(this.options.url)

    this.ws.onopen = () => {
      this.retryCount = 0
      this.startHeartbeat()
      // 连接就绪后发送队列中缓存的消息
      this.flushQueue()
      this.options.onOpen()
    }

    this.ws.onmessage = (event) => {
      // 过滤心跳 pong 响应
      try {
        const data = JSON.parse(event.data as string)
        if (data.type === 'pong') return
        this.options.onMessage(data)
      } catch {
        this.options.onMessage(event.data)
      }
    }

    this.ws.onerror = (event) => {
      this.options.onError(event)
    }

    this.ws.onclose = (event) => {
      this.stopHeartbeat()
      this.options.onClose(event)

      // 非正常关闭时自动重连
      if (!this.destroyed && event.code !== 1000) {
        this.scheduleReconnect()
      }
    }
  }

  private scheduleReconnect() {
    if (this.retryCount >= this.options.maxRetries) {
      console.error('WebSocket 重连次数已达上限，停止重连')
      return
    }

    // 指数退避：1s, 2s, 4s, 8s, 16s...
    const delay = this.options.retryDelay * Math.pow(2, this.retryCount)
    this.retryCount++
    console.log(`${delay}ms 后进行第 ${this.retryCount} 次重连...`)
    setTimeout(() => this.connect(), delay)
  }

  private startHeartbeat() {
    if (this.options.heartbeatInterval <= 0) return
    this.heartbeatTimer = setInterval(() => {
      if (this.ws?.readyState === WebSocket.OPEN) {
        this.ws.send(this.options.heartbeatMessage)
      }
    }, this.options.heartbeatInterval)
  }

  private stopHeartbeat() {
    if (this.heartbeatTimer !== null) {
      clearInterval(this.heartbeatTimer)
      this.heartbeatTimer = null
    }
  }

  private flushQueue() {
    while (this.messageQueue.length > 0) {
      const msg = this.messageQueue.shift()!
      this.ws?.send(msg)
    }
  }

  send(data: unknown) {
    const message = typeof data === 'string' ? data : JSON.stringify(data)

    if (this.ws?.readyState === WebSocket.OPEN) {
      this.ws.send(message)
    } else {
      // 连接未就绪时缓存消息，等连接建立后发送
      this.messageQueue.push(message)
    }
  }

  destroy() {
    this.destroyed = true
    this.stopHeartbeat()
    this.ws?.close(1000, 'Client destroyed')
    this.ws = null
  }

  get readyState(): number {
    return this.ws?.readyState ?? WebSocket.CLOSED
  }
}

Vue 3 集成

自定义 Composable

// composables/useWebSocket.ts
import { ref, onUnmounted, readonly } from 'vue'

interface UseWebSocketOptions {
  onMessage?: (data: unknown) => void
  onOpen?: () => void
  onClose?: (event: CloseEvent) => void
  immediate?: boolean
}

export function useWebSocket(url: string, options: UseWebSocketOptions = {}) {
  const ws = ref<WebSocket | null>(null)
  const isConnected = ref(false)
  const lastMessage = ref<unknown>(null)

  function connect() {
    if (ws.value?.readyState === WebSocket.OPEN) return

    ws.value = new WebSocket(url)

    ws.value.onopen = () => {
      isConnected.value = true
      options.onOpen?.()
    }

    ws.value.onmessage = (event) => {
      try {
        lastMessage.value = JSON.parse(event.data as string)
      } catch {
        lastMessage.value = event.data
      }
      options.onMessage?.(lastMessage.value)
    }

    ws.value.onclose = (event) => {
      isConnected.value = false
      options.onClose?.(event)
    }
  }

  function send(data: unknown) {
    const message = typeof data === 'string' ? data : JSON.stringify(data)
    if (ws.value?.readyState === WebSocket.OPEN) {
      ws.value.send(message)
    }
  }

  function disconnect() {
    ws.value?.close(1000)
    ws.value = null
    isConnected.value = false
  }

  if (options.immediate !== false) {
    connect()
  }

  onUnmounted(() => {
    disconnect()
  })

  return {
    isConnected: readonly(isConnected),
    lastMessage: readonly(lastMessage),
    connect,
    disconnect,
    send
  }
}

<!-- 组件中使用 -->
<script setup lang="ts">
import { useWebSocket } from '@/composables/useWebSocket'

const { isConnected, lastMessage, send } = useWebSocket('ws://localhost:8000/ws', {
  onMessage: (data) => {
    console.log('收到消息:', data)
  }
})
</script>

VueUse 的 useWebSocket

VueUse 提供了开箱即用的 useWebSocket，支持自动重连和心跳：

npm install @vueuse/core

import { useWebSocket } from '@vueuse/core'

const { status, data, send, close } = useWebSocket('ws://localhost:8000/ws', {
  autoReconnect: {
    retries: 5,
    delay: 1000,
    onFailed() {
      console.error('重连失败')
    }
  },
  heartbeat: {
    message: JSON.stringify({ type: 'ping' }),
    interval: 30000,
    pongTimeout: 5000
  }
})

VueUse useWebSocket 主要参数：

React 集成

自定义 Hook

// hooks/useWebSocket.ts
import { useEffect, useRef, useState, useCallback } from 'react'

interface UseWebSocketOptions {
  onMessage?: (data: unknown) => void
  onOpen?: () => void
  onClose?: (event: CloseEvent) => void
}

interface UseWebSocketReturn {
  isConnected: boolean
  lastMessage: unknown
  send: (data: unknown) => void
}

export function useWebSocket(
  url: string,
  options: UseWebSocketOptions = {}
): UseWebSocketReturn {
  const wsRef = useRef<WebSocket | null>(null)
  const [isConnected, setIsConnected] = useState(false)
  const [lastMessage, setLastMessage] = useState<unknown>(null)

  // 用 ref 保存 options 回调，避免 useEffect 依赖频繁变化
  const optionsRef = useRef(options)
  optionsRef.current = options

  useEffect(() => {
    const ws = new WebSocket(url)
    wsRef.current = ws

    ws.onopen = () => {
      setIsConnected(true)
      optionsRef.current.onOpen?.()
    }

    ws.onmessage = (event) => {
      let data: unknown
      try {
        data = JSON.parse(event.data as string)
      } catch {
        data = event.data
      }
      setLastMessage(data)
      optionsRef.current.onMessage?.(data)
    }

    ws.onclose = (event) => {
      setIsConnected(false)
      optionsRef.current.onClose?.(event)
    }

    // 组件卸载时正确关闭连接
    return () => {
      ws.close(1000)
      wsRef.current = null
    }
  }, [url]) // url 变化时重建连接

  const send = useCallback((data: unknown) => {
    const message = typeof data === 'string' ? data : JSON.stringify(data)
    if (wsRef.current?.readyState === WebSocket.OPEN) {
      wsRef.current.send(message)
    }
  }, [])

  return { isConnected, lastMessage, send }
}

// 组件中使用
function ChatRoom({ roomId }: { roomId: string }) {
  const { isConnected, lastMessage, send } = useWebSocket(
    `ws://localhost:8000/ws/room/${roomId}`,
    {
      onMessage: (data) => {
        console.log('新消息:', data)
      }
    }
  )

  return (
    <div>
      <span>状态: {isConnected ? '已连接' : '未连接'}</span>
      <button onClick={() => send({ type: 'chat', content: 'Hello' })}>
        发送
      </button>
    </div>
  )
}

消息格式设计

JSON 协议

约定 type + payload 结构，方便在客户端进行消息路由：

interface BaseMessage {
  type: string
  id?: string       // 可选消息 ID，用于请求-响应配对
  timestamp?: number
}

interface ChatMessage extends BaseMessage {
  type: 'chat'
  payload: {
    roomId: string
    content: string
    userId: string
  }
}

interface NotificationMessage extends BaseMessage {
  type: 'notification'
  payload: {
    level: 'info' | 'warning' | 'error'
    text: string
  }
}

type AppMessage = ChatMessage | NotificationMessage

// 消息分发器
function handleMessage(raw: unknown) {
  const message = raw as AppMessage
  switch (message.type) {
    case 'chat':
      handleChat(message.payload)
      break
    case 'notification':
      handleNotification(message.payload)
      break
    default:
      console.warn('未知消息类型:', message)
  }
}

二进制协议

使用 ArrayBuffer + DataView 构建紧凑的二进制帧：

// 协议定义：[消息类型(1字节)] [消息ID(4字节)] [数据长度(4字节)] [数据(N字节)]

function encodeMessage(type: number, id: number, data: Uint8Array): ArrayBuffer {
  const buffer = new ArrayBuffer(9 + data.byteLength)
  const view = new DataView(buffer)

  view.setUint8(0, type)          // 1 字节：消息类型
  view.setUint32(1, id)           // 4 字节：消息 ID（大端序）
  view.setUint32(5, data.byteLength) // 4 字节：数据长度

  // 写入数据
  const dataView = new Uint8Array(buffer, 9)
  dataView.set(data)

  return buffer
}

function decodeMessage(buffer: ArrayBuffer) {
  const view = new DataView(buffer)
  const type = view.getUint8(0)
  const id = view.getUint32(1)
  const dataLength = view.getUint32(5)
  const data = new Uint8Array(buffer, 9, dataLength)

  return { type, id, data }
}

// 接收二进制消息
ws.binaryType = 'arraybuffer'  // 默认为 'blob'，需要改成 'arraybuffer'

ws.onmessage = (event) => {
  if (event.data instanceof ArrayBuffer) {
    const { type, id, data } = decodeMessage(event.data)
    console.log(type, id, data)
  }
}

服务端（Python FastAPI）

基础 WebSocket 路由

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
import json

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            # 接收文本消息
            data = await websocket.receive_text()
            message = json.loads(data)

            # 回复消息
            await websocket.send_text(json.dumps({
                "type": "echo",
                "payload": message
            }))
    except WebSocketDisconnect:
        print("客户端断开连接")

广播模式

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List
import json

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def broadcast(self, message: dict):
        data = json.dumps(message)
        for connection in self.active_connections:
            await connection.send_text(data)

manager = ConnectionManager()

@app.websocket("/ws/broadcast")
async def broadcast_endpoint(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            message = json.loads(data)
            await manager.broadcast(message)
    except WebSocketDisconnect:
        manager.disconnect(websocket)

Room 模式

from collections import defaultdict
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
import json

app = FastAPI()

class RoomManager:
    def __init__(self):
        # room_id -> List[WebSocket]
        self.rooms: dict[str, list[WebSocket]] = defaultdict(list)

    async def join(self, room_id: str, websocket: WebSocket):
        await websocket.accept()
        self.rooms[room_id].append(websocket)

    def leave(self, room_id: str, websocket: WebSocket):
        self.rooms[room_id].remove(websocket)
        if not self.rooms[room_id]:
            del self.rooms[room_id]

    async def broadcast_to_room(self, room_id: str, message: dict):
        data = json.dumps(message)
        for connection in self.rooms.get(room_id, []):
            await connection.send_text(data)

room_manager = RoomManager()

@app.websocket("/ws/room/{room_id}")
async def room_endpoint(websocket: WebSocket, room_id: str):
    await room_manager.join(room_id, websocket)
    try:
        while True:
            data = await websocket.receive_text()
            message = json.loads(data)
            await room_manager.broadcast_to_room(room_id, message)
    except WebSocketDisconnect:
        room_manager.leave(room_id, websocket)

最佳实践

连接状态管理

在发送消息前始终检查 readyState，避免向未就绪的连接发送消息：

function safeSend(ws: WebSocket, data: unknown) {
  if (ws.readyState !== WebSocket.OPEN) {
    console.warn(`连接状态为 ${ws.readyState}，消息发送失败`)
    return false
  }
  ws.send(typeof data === 'string' ? data : JSON.stringify(data))
  return true
}

请求-响应模式

为每条消息分配 ID，实现类 RPC 的请求-响应配对：

class WebSocketRPC {
  private ws: WebSocket
  private pending = new Map<string, (response: unknown) => void>()

  constructor(url: string) {
    this.ws = new WebSocket(url)
    this.ws.onmessage = (event) => {
      const message = JSON.parse(event.data as string)
      if (message.id && this.pending.has(message.id)) {
        const resolve = this.pending.get(message.id)!
        this.pending.delete(message.id)
        resolve(message.payload)
      }
    }
  }

  call(type: string, payload: unknown): Promise<unknown> {
    return new Promise((resolve, reject) => {
      const id = crypto.randomUUID()
      this.pending.set(id, resolve)

      // 超时处理
      setTimeout(() => {
        if (this.pending.has(id)) {
          this.pending.delete(id)
          reject(new Error(`请求 ${id} 超时`))
        }
      }, 10000)

      this.ws.send(JSON.stringify({ type, id, payload }))
    })
  }
}

踩坑与注意事项

跨域配置

WebSocket 握手基于 HTTP，受同源策略约束。服务端需要验证 Origin 请求头：

# FastAPI 允许指定来源
from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:5173"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Nginx 代理 WebSocket 必须添加 Upgrade 头

普通 Nginx 反向代理不支持 WebSocket，必须添加以下配置：

location /ws {
    proxy_pass http://backend:8000;
    proxy_http_version 1.1;

    # 这两行是关键，缺少任意一行都会导致 WebSocket 升级失败
    proxy_set_header Upgrade $http_upgrade;
    proxy_set_header Connection "upgrade";

    proxy_set_header Host $host;
    proxy_set_header X-Real-IP $remote_addr;

    # WebSocket 长连接，需要适当增大超时时间
    proxy_read_timeout 3600s;
    proxy_send_timeout 3600s;
}

连接数限制

浏览器对同一域名的 WebSocket 连接数有限制（通常约 256 个），服务器操作系统对文件描述符数量也有限制。大量并发连接场景需要：

• 调整系统文件描述符上限（ulimit -n）
• 使用连接池或复用单个连接（multiplexing）
• 考虑使用 WebSocket 网关（如 APISIX、Nginx Plus）

页面隐藏时的连接行为

部分浏览器会在页面进入后台时降低 WebSocket 的优先级甚至断开连接，需要监听 visibilitychange 事件重连：

document.addEventListener('visibilitychange', () => {
  if (document.visibilityState === 'visible') {
    if (ws.readyState === WebSocket.CLOSED) {
      // 重新建立连接
      reconnect()
    }
  }
})

避免在 React StrictMode 下重复建立连接

React StrictMode 在开发环境中会刻意挂载组件两次，导致 useEffect 触发两次，产生两个 WebSocket 连接。确保 cleanup 函数正确关闭连接：

useEffect(() => {
  const ws = new WebSocket(url)
  // ...

  // cleanup 函数必须关闭连接
  return () => {
    ws.close(1000)
  }
}, [url])

常见陷阱

陷阱：WebSocket 连接在移动端或弱网下频繁断开

现象： 移动端 App 切换到后台或网络抖动后，WebSocket 静默断开，服务端无感知，消息丢失。
原因： 大多数代理、负载均衡器（Nginx 默认 60s）和移动 OS 会在空闲超时后断开连接，且不发送 close 帧。
解决： 客户端实现心跳机制（每 30s 发送 ping 消息），服务端响应 pong；同时实现断线自动重连（指数退避）：

let reconnectDelay = 1000;
function connect() {
  const ws = new WebSocket(url);
  ws.onclose = () => {
    setTimeout(connect, reconnectDelay);
    reconnectDelay = Math.min(reconnectDelay * 2, 30000);
  };
  ws.onopen = () => { reconnectDelay = 1000; };
}

陷阱：跨域 WebSocket 连接被阻止

现象： 浏览器控制台报 WebSocket connection to 'wss://...' failed，HTTP 握手返回 403。
原因： WebSocket 握手是 HTTP 请求，服务端没有校验 Origin 头导致拒绝，或反向代理未转发 Upgrade: websocket 请求头。
解决： 服务端配置允许的 Origin，Nginx 配置添加：

location /ws {
    proxy_pass http://backend;
    proxy_http_version 1.1;
    proxy_set_header Upgrade $http_upgrade;
    proxy_set_header Connection "upgrade";
}

陷阱：发送大量消息时内存无限增长

现象： 高频发送消息时内存持续增长，最终崩溃。
原因： 发送速度超过网络吞吐时，消息积压在 bufferedAmount 缓冲区，若不检查缓冲状态就持续调用 send()，缓冲区无限膨胀。
解决： 发送前检查 ws.bufferedAmount，缓冲区积压超阈值时暂停发送或丢弃低优先级消息：

if (ws.bufferedAmount < 1024 * 64) {
  ws.send(data);
}

参见

Axios完全指南
Vue3入门