跳到主要内容

WebSocket 实时通信

WebSocket 是 Node.js 服务的核心职责：承接传统后端不擅长的海量长连接场景，为 Web / App / 小程序提供实时推送能力。

技术选型​

Socket.IO（@nestjs/platform-socket.io）：自动降级（WebSocket → 长轮询）、房间/命名空间、ack 机制、生态成熟。

pnpm add @nestjs/websockets @nestjs/platform-socket.io

网关定义​

网关是 WebSocket 入口，按业务域划分，一个业务域一个网关：

// modules/notification/notification.gateway.ts
import {
  ConnectedSocket,
  MessageBody,
  SubscribeMessage,
  WebSocketGateway,
  WebSocketServer,
} from '@nestjs/websockets';
import { Server, Socket } from 'socket.io';

@WebSocketGateway({
  namespace: 'notification',
  cors: { origin: false },
})
export class NotificationGateway {
  @WebSocketServer()
  server: Server;

  @SubscribeMessage('notification:subscribe')
  handleSubscribe(
    @MessageBody() data: SubscribeTopicDto,
    @ConnectedSocket() client: Socket,
  ) {
    return { code: 0, message: 'ok', data: { topic: data.topic } };
  }
}
  • 网关注册在所属模块的 providers 中（未被模块引用的网关不会实例化）
  • 命名空间按业务域隔离（/notification、/device），避免单一命名空间事件爆炸
  • @WebSocketGateway() 第二参数可透传任意 Socket.IO Server 选项

连接鉴权​

令牌在握手阶段校验，非法连接立即断开。鉴权逻辑复用全局 JWT 能力，不单独造轮子：

// modules/notification/notification.gateway.ts
export class NotificationGateway implements OnGatewayConnection, OnGatewayDisconnect {
  constructor(private readonly jwtService: JwtService) {}

  async handleConnection(client: Socket): Promise<void> {
    try {
      const token = client.handshake.auth.token as string;
      const payload = await this.jwtService.verifyAsync<AuthPayload>(token);
      client.data.userId = payload.sub;
      client.data.tenantId = payload.tenantId;
      await client.join(`tenant:${payload.tenantId}`);
    } catch {
      client.disconnect(true);
    }
  }

  handleDisconnect(client: Socket): void {
  }
}

客户端连接：

import { io } from 'socket.io-client';

const socket = io('wss://api.example.com/notification', {
  auth: { token },
  transports: ['websocket'],
});
注意
  • 令牌禁止通过 URL query 传递（会进入网关与代理的访问日志）
  • 连接建立后令牌过期不强制踢线，但后续敏感事件需重新校验；长连接续期由客户端重连完成

消息收发​

@SubscribeMessage('notification:mark-read')
handleMarkRead(@MessageBody() data: MarkReadDto) {
  return { code: 0, message: 'ok', data: { id: data.id } };
}
// 客户端
socket.emit('notification:mark-read', { id: 123 }, (res) => {
  console.log(res.code);
});

房间与广播域​

房间是推送的最小寻址单元，命名规则与订阅事件对齐：

房间名覆盖范围加入时机
user:{userId}单用户全端连接建立时
tenant:{tenantId}租户内全部用户连接建立时
topic:{topic}主题订阅者客户端订阅事件后
  • 加入/退出房间必须通过服务端方法执行，客户端不能直接声明房间
  • 敏感房间（如 tenant:）加入前校验用户归属，禁止任意加入

多实例广播​

单实例 server.emit 只能触达本实例连接。生产环境多实例部署时，必须接入 Redis 适配器，让广播跨实例生效：

pnpm add @socket.io/redis-adapter ioredis
// main.ts
import { IoAdapter } from '@nestjs/platform-socket.io';
import { createAdapter } from '@socket.io/redis-adapter';
import { createClient } from 'redis';

class RedisIoAdapter extends IoAdapter {
  private adapterConstructor: ReturnType<typeof createAdapter>;

  async connectToRedis(): Promise<void> {
    const pubClient = createClient({ url: process.env.REDIS_URL });
    const subClient = pubClient.duplicate();
    await Promise.all([pubClient.connect(), subClient.connect()]);
    this.adapterConstructor = createAdapter(pubClient, subClient);
  }

  createIOServer(port: number, options?: ServerOptions) {
    const server = super.createIOServer(port, options);
    server.adapter(this.adapterConstructor);
    return server;
  }
}

// bootstrap 中
const redisIoAdapter = new RedisIoAdapter(app);
await redisIoAdapter.connectToRedis();
app.useWebSocketAdapter(redisIoAdapter);
配套要求：网关 / LB 必须开启 Sticky Session

Redis Adapter 只解决“跨实例消息可达”，不解决“连接归属”。Socket.IO 握手由多次 HTTP 请求组成（polling → upgrade），若网关轮询分发，握手请求落到不同实例会直接失败。多实例部署时必须同时满足：

  1. 网关 / 负载均衡 Sticky Session：按客户端 IP 或 Cookie 将同一客户端固定到同一实例（负载均衡层配置见 网关接入 · WebSocket 路由要点）
  2. Redis Adapter：跨实例广播寻址（本节）

心跳与超时​

Socket.IO 内置 ping/pong 心跳，生产环境按弱网环境调整：

@WebSocketGateway({
  pingInterval: 25000,
  pingTimeout: 20000,
  connectTimeout: 10000,
})
  • 移动端建议 pingInterval ≤ 30s，避免运营商 NAT 会话老化断连
  • 客户端实现指数退避重连（Socket.IO Client 默认支持），重连成功后重新订阅房间

与传统后端的协作时序​

实时数据的权威来源仍是传统后端，Node.js 只做分发：

边界红线
  • Node.js 服务不接收客户端的写操作指令直接落库；客户端的写请求走传统后端 REST，变更结果经 Kafka 事件回流为推送
  • 离线消息不在 WebSocket 通道补发，归传统后端的站内信/APNs 通道