Nest.js WebSocket网关结合Redis Broker推送失效问题排查与解决

问题背景 #

单实例环境下，Nest.js的WebSocket网关可正常推送消息，但部署到负载均衡后的多实例环境时，客户端可能连接到与消息推送端不同的实例，导致推送失效。

尝试通过Redis Pub/Sub作为消息代理实现跨实例消息同步：消息能正常存入Redis并被订阅端读取，但调用推送方法时，目标房间始终显示无连接客户端——尽管已在handleConnection中确认客户端成功加入对应房间。调整Gateway作用域为默认值后问题仍存在。

初始实现代码 #

SocketIOAdapter #

export class SocketIOAdapter extends IoAdapter {
  constructor(private app: INestApplicationContext, private configService: ConfigService) {
    super(app);
  }

  createIOServer(port: number, options: ServerOptions) {
    const clientPort = parseInt(this.configService.getOrThrow("PORT"));

    const cors = {
      origin: [
        `http://localhost:${clientPort}`,
        new RegExp(`/^http:\/\/192\.168\.1\.([1-9]|[1-9]\d):${clientPort}$/`),
      ],
    };

    const optionsWithCORS: ServerOptions = {
      ...options,
      cors,
    };

    const server: Server = super.createIOServer(port, optionsWithCORS);

    const orderRepository = this.app.get(OrderRepository);
    server
      .of("/orders")
      .use(createTokenMiddleware(orderRepository));

    return server;
  }
}

const createTokenMiddleware =
  (orderRepository: OrderRepository) =>
  async (socket: Socket, next) => {
    // 此处使用orderRepository执行逻辑
    next();
  } catch {
    next(new Error("FORBIDDEN"));
  }
};

OrderGateway #

@WebSocketGateway({
  namespace: "/orders",
})
@Injectable()
export class OrderGateway implements OnGatewayInit, OnGatewayConnection, OnGatewayDisconnect {
  private readonly logger = new Logger(OrderGateway.name);

  @WebSocketServer() io: Namespace;

  afterInit(): void {
    this.logger.log("Websocket Gateway initialized.");
  }

  async handleConnection(client: Socket) {
    const sockets = this.io.sockets;

    // 此处逻辑确定客户端需要加入的房间
    const roomsToJoin = [...]
    await client.join(roomsToJoin);
  }

  async handleDisconnect(client: Socket) {
    this.logger.log(`Disconnected socket id: ${client.id}`);
  }

  public emitOrderStatusChangeNotification(order: OrderDTO) {
    this.io
      .to("根据order生成的roomId")
      .emit("order_status_changed", JSON.stringify(order));
  }
}

尝试的Redis Pub/Sub方案 #

消息发布服务 #

@Injectable()
export class NotificationPublisherService {
  constructor(@Inject("ORDER_NOTIFICATION_SERVICE") private client: ClientProxy) {}

  async publishEvent(order: OrderDTO) {
    console.log("将消息发送到redis");
    this.client.emit(Constants.notificationEventName, order);
  }
}

消息订阅控制器 #

@Controller()
export class NotificationSuscriberController {
  private readonly logger = new Logger(NotificationSuscriberController.name);

  constructor(private readonly orderGateway: OrderGateway) {}

  @EventPattern(Constants.notificationEventName)
  async handleOrderStatusChangeEvent(order: OrderDTO) {
    try {
      this.orderGateway.emitOrderStatusChangeNotification(order);
    } catch (err) {
      this.logger.log("发送通知出错");
    }
  }
}

将原有的emitOrderStatusChangeNotification调用替换为publishEvent后，消息能正常在Redis中收发，但推送时目标房间无客户端连接的问题未解决。

最终解决方案 #

通过扩展IoAdapter并集成Redis适配器，实现跨实例的WebSocket房间状态同步，修改后的代码如下：

修改后的SocketIOAdapter #

export class SocketIOAdapter extends IoAdapter {
  private adapterConstructor: ReturnType<typeof createAdapter>;
  constructor(private app: INestApplicationContext, private configService: ConfigService) {
    super(app);
  }

  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: ServerOptions) {
    const clientPort = parseInt(this.configService.getOrThrow("PORT"));

    const cors = {
      origin: [
        `http://localhost:${clientPort}`,
        new RegExp(`/^http:\/\/192\.168\.1\.([1-9]|[1-9]\d):${clientPort}$/`),
      ],
    };

    const optionsWithCORS: ServerOptions = {
      ...options,
      cors,
    };

    const server: Server = super.createIOServer(port, optionsWithCORS);

    const orderRepository = this.app.get(OrderRepository);
    server
      .adapter(this.adapterConstructor)
      .of(`/orders`)
      .use(createTokenMiddleware(orderRepository));

    return server;
  }
}

const createTokenMiddleware =
  (orderRepository: OrderRepository) =>
  async (socket: Socket, next) => {
    // 此处使用orderRepository执行逻辑
    next();
  } catch {
    next(new Error("FORBIDDEN"));
  }
};

main.ts配置 #

const redisIoAdapter = new SocketIOAdapter(app, configService);
await redisIoAdapter.connectToRedis();

通过上述修改，所有实例共享Redis作为WebSocket的状态存储，跨实例的消息推送可正常触达目标房间的客户端。

内容的提问来源于stack exchange，提问作者Juan Chaves