nestjs-websocket

NestJS WebSocket 集成

深入了解：使用 technology: nestjs 的 mcp__documentation__fetch_docs 获取全面的文档。

网关设置

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

@WebSocketGateway({
  cors: {
    origin: process.env.CORS_ORIGINS?.split(',') || [],
    credentials: true,
  },
  namespace: '/events',
})
export class EventsGateway implements OnGatewayConnection, OnGatewayDisconnect {
  @WebSocketServer()
  server: Server;

  handleConnection(client: Socket) {
    console.log(`Client connected: ${client.id}`);
  }

  handleDisconnect(client: Socket) {
    console.log(`Client disconnected: ${client.id}`);
  }

  @SubscribeMessage('message')
  handleMessage(
    @ConnectedSocket() client: Socket,
    @MessageBody() payload: { room: string; data: any },
  ) {
    this.server.to(payload.room).emit('message', payload.data);
    return { event: 'message', data: 'Message sent' };
  }
}

模块注册

// events.module.ts
import { Module } from '@nestjs/common';
import { EventsGateway } from './events.gateway';

@Module({
  providers: [EventsGateway],
  exports: [EventsGateway],
})
export class EventsModule {}

// app.module.ts
@Module({
  imports: [EventsModule],
})
export class AppModule {}

身份验证守卫

// ws-auth.guard.ts
import { CanActivate, ExecutionContext, Injectable } from '@nestjs/common';
import { WsException } from '@nestjs/websockets';
import { Socket } from 'socket.io';
import { JwtService } from '@nestjs/jwt';

@Injectable()
export class WsAuthGuard implements CanActivate {
  constructor(private jwtService: JwtService) {}

  async canActivate(context: ExecutionContext): Promise<boolean> {
    const client: Socket = context.switchToWs().getClient();
    const token = client.handshake.auth?.token ||
                  client.handshake.headers?.authorization?.split(' ')[1];

    if (!token) {
      throw new WsException('Unauthorized');
    }

    try {
      const payload = await this.jwtService.verifyAsync(token);
      client.data.user = payload;
      return true;
    } catch {
      throw new WsException('Invalid token');
    }
  }
}

// Usage in gateway
@UseGuards(WsAuthGuard)
@SubscribeMessage('protected-event')
handleProtectedEvent(@ConnectedSocket() client: Socket) {
  const user = client.data.user;
  return { user };
}

房间管理

@WebSocketGateway()
export class RoomsGateway {
  @WebSocketServer()
  server: Server;

  @SubscribeMessage('join-room')
  handleJoinRoom(
    @ConnectedSocket() client: Socket,
    @MessageBody() roomId: string,
  ) {
    client.join(roomId);
    client.to(roomId).emit('user-joined', { id: client.id });
    return { event: 'joined', room: roomId };
  }

  @SubscribeMessage('leave-room')
  handleLeaveRoom(
    @ConnectedSocket() client: Socket,
    @MessageBody() roomId: string,
  ) {
    client.leave(roomId);
    client.to(roomId).emit('user-left', { id: client.id });
  }

  // Broadcast to room from service
  broadcastToRoom(roomId: string, event: string, data: any) {
    this.server.to(roomId).emit(event, data);
  }

  // Get clients in room
  async getClientsInRoom(roomId: string): Promise<string[]> {
    const sockets = await this.server.in(roomId).fetchSockets();
    return sockets.map((s) => s.id);
  }
}

异常处理

// ws-exception.filter.ts
import { Catch, ArgumentsHost } from '@nestjs/common';
import { BaseWsExceptionFilter, WsException } from '@nestjs/websockets';
import { Socket } from 'socket.io';

@Catch()
export class WsExceptionFilter extends BaseWsExceptionFilter {
  catch(exception: unknown, host: ArgumentsHost) {
    const client: Socket = host.switchToWs().getClient();

    const error = exception instanceof WsException
      ? exception.getError()
      : { message: 'Internal server error' };

    client.emit('error', {
      status: 'error',
      message: typeof error === 'string' ? error : (error as any).message,
    });
  }
}

// Apply globally or per gateway
@UseFilters(new WsExceptionFilter())
@WebSocketGateway()
export class EventsGateway {}

使用 Redis 适配器进行扩展

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

export 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?: any) {
    const server = super.createIOServer(port, options);
    server.adapter(this.adapterConstructor);
    return server;
  }
}

// Bootstrap
async function bootstrap() {
  const app = await NestFactory.create(AppModule);

  const redisIoAdapter = new RedisIoAdapter(app);
  await redisIoAdapter.connectToRedis();
  app.useWebSocketAdapter(redisIoAdapter);

  await app.listen(3000);
}

事件监听器

@Component
@RequiredArgsConstructor
@Slf4j
export class WebSocketEventListener {
  constructor(private messagingTemplate: SimpMessagingTemplate) {}

  @EventListener
  handleWebSocketConnectListener(event: SessionConnectedEvent) {
    log.info('New WebSocket connection established');
  }

  @EventListener
  handleWebSocketDisconnectListener(event: SessionDisconnectEvent) {
    const headerAccessor = StompHeaderAccessor.wrap(event.getMessage());
    const username = headerAccessor.getSessionAttributes().get('username');

    if (username != null) {
      log.info(`User disconnected: ${username}`);
      const leaveMessage = ChatMessage.builder()
        .type(ChatMessage.MessageType.LEAVE)
        .sender(username)
        .build();
      messagingTemplate.convertAndSend('/topic/public', leaveMessage);
    }
  }
}

在服务中注入网关

// notifications.service.ts
@Injectable()
export class NotificationsService {
  constructor(private eventsGateway: EventsGateway) {}

  async notifyUser(userId: string, notification: Notification) {
    this.eventsGateway.server
      .to(`user:${userId}`)
      .emit('notification', notification);
  }

  async broadcastAnnouncement(message: string) {
    this.eventsGateway.server.emit('announcement', { message });
  }
}

反模式

反模式	不足之处	正确做法
不使用命名空间	所有事件混杂在一起	使用命名空间进行隔离
连接时缺少身份验证	未经授权的访问	实现 WsAuthGuard
不清理房间	内存泄漏	处理断开连接事件
在处理程序中执行同步的繁重操作	阻塞事件循环	正确使用 async/await
生产环境中缺少 Redis 适配器	无法水平扩展	使用 Redis 适配器

快速故障排除

症状	原因	解决方法
CORS 错误	缺少 CORS 配置	为网关添加 cors 选项
身份验证始终失败	握手中没有令牌	检查客户端身份验证设置
未收到消息	命名空间错误	确认命名空间一致
房间广播失败	客户端不在房间内	确认已调用 join-room
扩展问题	没有 Redis 适配器	实现 RedisIoAdapter

生产环境检查清单

• 正确配置 CORS
• JWT 身份验证守卫
• 用于水平扩展的 Redis 适配器
• 用于错误处理的异常过滤器
• 正确处理断开连接
• 对消息进行速率限制
• 配置心跳/ping
• 记录连接/断开连接日志

不适用本技能的情况

• React WebSocket 客户端 → 使用 react-websocket
• Spring WebSocket → 使用 spring-websocket
• 原始 HTTP 长轮询 → 使用标准控制器

参考文档

• NestJS 核心
• JWT 身份验证