WebSocket网关

为什么需要 WebSocket

HTTP 是”你问我答”的模式——客户端发请求，服务端返回响应。但有些场景需要服务端主动推送消息给客户端：聊天室、实时通知、协作编辑、股票行情。

轮询可以解决，但太浪费资源。WebSocket 就是为这个问题而生的——建立一次连接后，双方可以随时互发消息，不需要反复建立连接。

NestJS 把 WebSocket 封装成了网关（Gateway），写法跟控制器很像，上手很快。

安装

NestJS 内置支持两种 WebSocket 库：socket.io 和 ws。

socket.io 功能丰富（自动重连、房间、命名空间），但体积大一些。ws 更轻量、更快，但功能少。

一般推荐 socket.io，除非你有明确的性能需求。

npm install @nestjs/websockets @nestjs/platform-socket.io

创建网关

网关就是一个类，加上 @WebSocketGateway() 装饰器：

import { WebSocketGateway, SubscribeMessage, MessageBody } from '@nestjs/websockets';

@WebSocketGateway()
export class EventsGateway {
  @SubscribeMessage('events')
  handleEvent(@MessageBody() data: string): string {
    return data;
  }
}

@SubscribeMessage('events') 监听客户端发来的 events 消息。@MessageBody() 取出消息体。

把网关注册到模块里：

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

配置命名空间和选项

@WebSocketGateway 支持传 socket.io 的配置项：

@WebSocketGateway(8080, {
  namespace: 'events',
  cors: true,
  transports: ['websocket', 'polling'],
})
export class EventsGateway {}

namespace 用来隔离不同的通信频道。客户端连接时需要指定命名空间：

const socket = io('http://localhost:3000/events');

发送消息给客户端

方式一：直接 return

最简单的做法——handler 的返回值会自动作为确认（acknowledgment）发回给客户端：

@SubscribeMessage('events')
handleEvent(@MessageBody() data: string): string {
  // 客户端收到的确认就是 data 原样返回
  return data;
}

客户端这样接收：

socket.emit('events', { name: 'Nest' }, (response) => {
  console.log(response); // { name: 'Nest' }
});

方式二：指定事件名发送

如果不想用 acknowledgment，而是想主动 emit 一个事件给客户端，返回一个 WsResponse 对象：

import { WsResponse } from '@nestjs/websockets';

@SubscribeMessage('events')
handleEvent(@MessageBody() data: unknown): WsResponse<unknown> {
  return { event: 'events-response', data: { message: '收到你的消息了' } };
}

客户端监听 events-response 事件：

socket.on('events-response', (data) => {
  console.log(data); // { message: '收到你的消息了' }
});

方式三：用 socket 实例直接发

注入 @ConnectedSocket() 拿到 socket 实例，想怎么发就怎么发：

import { ConnectedSocket } from '@nestjs/websockets';
import { Socket } from 'socket.io';

@SubscribeMessage('events')
handleEvent(
  @MessageBody() data: string,
  @ConnectedSocket() client: Socket,
): void {
  // 给当前客户端发消息
  client.emit('custom-event', { result: '处理完毕' });

  // 广播给所有客户端
  this.server.emit('broadcast-event', { info: '所有人注意' });
}

异步和 Observable 响应

handler 支持 async/await：

@SubscribeMessage('events')
async handleEvent(@MessageBody() data: string): Promise<string> {
  const result = await this.someService.process(data);
  return result;
}

也可以返回 Observable，会逐个发送流中的数据：

@SubscribeMessage('events')
handleEvent(@MessageBody() data: unknown): Observable<WsResponse<number>> {
  const event = 'events';
  const response = [1, 2, 3];

  return from(response).pipe(
    map(data => ({ event, data })),
  );
}

客户端会收到 3 次消息，分别是 1、2、3。

生命周期钩子

网关有三个生命周期钩子，跟 HTTP 的生命周期类似：

钩子	接口	触发时机
afterInit	OnGatewayInit	服务器初始化完成后
handleConnection	OnGatewayConnection	客户端连接时
handleDisconnect	OnGatewayDisconnect	客户端断开时

import {
  WebSocketGateway,
  OnGatewayInit,
  OnGatewayConnection,
  OnGatewayDisconnect,
} from '@nestjs/websockets';
import { Logger } from '@nestjs/common';
import { Socket, Server } from 'socket.io';

@WebSocketGateway()
export class EventsGateway
  implements OnGatewayInit, OnGatewayConnection, OnGatewayDisconnect
{
  private readonly logger = new Logger(EventsGateway.name);

  @WebSocketServer()
  server: Server;

  afterInit(server: Server) {
    this.logger.log('WebSocket 网关初始化完成');
  }

  handleConnection(client: Socket) {
    this.logger.log(`客户端连接: ${client.id}`);
  }

  handleDisconnect(client: Socket) {
    this.logger.log(`客户端断开: ${client.id}`);
  }
}

@WebSocketServer() 装饰器把 socket.io 的 Server 实例注入到属性上。你可以在任何地方用它来广播消息。

广播消息

一个常见需求是：某个客户端发了消息，要推送给所有人。

@SubscribeMessage('chat')
handleChat(
  @MessageBody() message: string,
  @ConnectedSocket() client: Socket,
): void {
  // 广播给所有人（包括发送者自己）
  this.server.emit('chat', {
    from: client.id,
    message,
  });
}

如果不想发给发送者自己：

client.broadcast.emit('chat', {
  from: client.id,
  message,
});

房间（Room）

socket.io 的房间功能让你可以给一组客户端发消息：

@SubscribeMessage('joinRoom')
handleJoinRoom(
  @MessageBody() room: string,
  @ConnectedSocket() client: Socket,
): void {
  client.join(room);
  client.emit('joined', `你加入了房间 ${room}`);
}

@SubscribeMessage('roomMessage')
handleRoomMessage(
  @MessageBody() data: { room: string; message: string },
  @ConnectedSocket() client: Socket,
): void {
  // 只发给房间里的所有人
  this.server.to(data.room).emit('roomMessage', {
    from: client.id,
    message: data.message,
  });
}

WebSocket 适配器

默认用 socket.io 的适配器 IoAdapter。但你可能需要换适配器，比如多实例部署时需要 Redis 来同步消息。

用 ws 库替代 socket.io

如果你觉得 socket.io 太重，可以换成 ws：

npm install @nestjs/platform-ws

import { WsAdapter } from '@nestjs/platform-ws';

const app = await NestFactory.create(AppModule);
app.useWebSocketAdapter(new WsAdapter(app));

Redis 适配器（多实例部署）

如果你的服务部署了多个实例，客户端 A 连到实例 1，客户端 B 连到实例 2，它们之间的消息就传不过去了。因为每个实例各自维护自己的连接。

解决办法是用 Redis 做消息中转：

npm install redis socket.io @socket.io/redis-adapter

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: 'redis://localhost:6379' });
    const subClient = pubClient.duplicate();

    await Promise.all([pubClient.connect(), subClient.connect()]);
    this.adapterConstructor = createAdapter(pubClient, subClient);
  }

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

在 main.ts 里启用：

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

异常处理

网关里的异常处理跟控制器一样——可以用异常过滤器。抛出的异常会自动发给客户端：

@SubscribeMessage('events')
handleEvent(@MessageBody() data: string): void {
  if (!data) {
    throw new WsException('数据不能为空');
  }
  // ...
}

WsException 是 WebSocket 专用的异常类，从 @nestjs/websockets 导入。

守卫和管道

网关跟控制器一样，支持守卫、管道、拦截器。用法完全一致：

@WebSocketGateway()
@UseGuards(AuthGuard)
export class EventsGateway {
  @SubscribeMessage('events')
  @UsePipes(new ValidationPipe())
  handleEvent(@MessageBody() data: CreateEventDto): void {
    // data 已经经过验证了
  }
}

小结

• 网关用 @WebSocketGateway() 装饰，注册到模块的 providers 里
• @SubscribeMessage() 监听消息，@MessageBody() 取数据
• 返回数据自动作为确认发回，返回 WsResponse 可以指定事件名
• 三个生命周期钩子：afterInit、handleConnection、handleDisconnect
• 多实例部署用 Redis 适配器同步消息
• 网关支持守卫、管道、拦截器，用法跟控制器一致