实时在线人数展示，WebSocket or SSE ?

前言

WebSocket是一种在单个TCP连接上提供全双工通信的网络协议。它允许客户端和服务器之间进行实时数据传输，而无需进行多次HTTP请求。WebSocket协议通过在客户端和服务器之间建立持久连接来实现双向通信，从而使得服务器可以主动向客户端推送数据，而不仅仅是响应客户端的请求。

WebSocket通常用于实现实时聊天应用、在线游戏、实时数据更新等需要实时通信的场景。通过WebSocket，客户端和服务器可以实时交换数据，而无需频繁的轮询或长轮询来获取更新。

SSE（Server-Sent Events，服务器发送事件）是一种用于在客户端和服务器之间建立单向实时通信的技术。与WebSocket不同，SSE是基于HTTP协议的，使用标准的HTTP连接，因此不需要像WebSocket那样建立新的协议，允许服务器向客户端推送事件流，而客户端则可以通过监听这些事件来接收实时数据更新。

在使用SSE时，服务器端需要设置特定的HTTP响应头（如Content-Type: text/event-stream）来指示客户端接收事件流。客户端则可以通过EventSource API来建立与服务器的连接，并通过监听message事件来处理接收到的事件流数据。SSE通常用于实时通知、实时数据更新、实时监控等需要服务器主动向客户端推送数据的场景。相比于WebSocket，SSE更适用于单向数据推送的应用场景，如新闻更新、实时股票报价等。

上一篇文章里我们模拟实现了 Redis 存储在线人数，文章最后提供了几种前端实时展示在线人数的方案，本文介绍其中的 WebSocket 和 SSE 的实现方案。

WebSocket

Nest.js 官方有 WebSocket 相关的使用文档，详情可见：https://docs.nestjs.com/websockets/gateways

“

在 Nest 中，网关只是一个用 @WebSocketGateway() 装饰器注释的类。从技术上讲，网关与平台无关，这使得它们在创建适配器后与任何 WebSockets 库兼容。有两个开箱即用的 WS 平台受支持：socket.io 和 ws。

我们这里使用 socket.io

后端实现

nest 项目下安装依赖：

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

nest 项目下新建 events 存放网关模块。

// events/events.modules.ts

import { Module } from '@nestjs/common';

import { OnlineModule } from '../online/online.module';
import { EventGateway } from './events.gateway';

@Module({
  imports: [OnlineModule],
  providers: [EventGateway],
  exports: [EventGateway],
})
export class EventsModule {}

// events/events.gateway.ts

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

import { OnlineService } from '../online/online.service';

@WebSocketGateway({
  cors: {
    origin: 'http://localhost:3000',
  },
})
export class EventGateway
  implements OnGatewayInit, OnGatewayConnection, OnGatewayDisconnect
{
  @WebSocketServer()
  server: Server;
  constructor(private readonly onlineService: OnlineService) {}

  async getOnlineUsers() {
    const { count } = await this.onlineService.getOnlineUserCount();
    this.server.emit('onlineUsers', count);
  }

  async afterInit() {
    this.server.on('connection', () => {
      this.getOnlineUsers();
    });
  }

  handleConnection(@ConnectedSocket() client: Socket) {
    const userId = client.handshake.headers['user-id'] as string;
    if (!userId) return;
    this.onlineService.addUser(userId);
    this.getOnlineUsers();
  }

  handleDisconnect(@ConnectedSocket() client: Socket) {
    const userId = client.handshake.headers['user-id'] as string;
    if (!userId) return;
    this.onlineService.removeUser(userId);
    this.getOnlineUsers();
  }
}

简单解释一下上面的代码。

@WebSocketGateway 装饰器注释的类表示网关，其中参数 cors 配置前端项目的地址。函数 getOnlineUsers() 调用上篇文章中写好的 OnlineService 中获取所有在线人数的方法，通过 emit 方法将数据发布出去，自定义的事件名为 onlineUsers。

EventGateway 继承了 OnGatewayInit, OnGatewayConnection, OnGatewayDisconnect。那就可以直接使用默认的方法，afterInit 表示初始化阶段会调用一次，在执行的时候判断只有连接阶段才会调用 getOnlineUsers() 获取在线人数。handleConnection 表示 websocket 连接上时候调用，用户登录上线连接上websocket，wss 协议中传输 userId，我们可以将数据放在请求头上进行传输。handleDisconnect 表示在断开连接的时候调用。

最后需要在 app.module 中引入 EventsModule，并在 imports 中注册。

前端实现

前端项目下安装依赖：

pnpm add socket.io-client

import io from "socket.io-client";
import { ref } from "vue";

function useSocket(userId: string) {
  const socket = io("ws://localhost:3001", {
    transportOptions: {
      polling: {
        extraHeaders: {
          "User-Id": userId,
        },
      },
    },
  });
  return {
    socket,
  };
}

export function useOnline() {
  const onlineUserCount = ref(0);

  const userId = `${Math.random()}`;
  const { socket } = useSocket(userId);

  function watchOnlineUsers() {
    socket.on("onlineUsers", (count: number) => {
      onlineUserCount.value = count;
    });
  }

  const leaveGame = () => socket.close();

  return {
    onlineUserCount,
    watchOnlineUsers,
    leaveGame,
  };
}

其中 userId 需要获取当前登录用户的 userId，这里代码仅作为模拟使用，这样每次刷新的时候都会随机生成一个 userId 都会导致 onlineUsers 事件的订阅。

SSE

Nest.js 官方有 SSE 相关的使用文档，详情可见: https://docs.nestjs.com/techniques/server-sent-events

“

要在路由（在控制器类中注册的路由）上启用服务器发送的事件，请使用 @Sse() 装饰器注释方法处理程序。

相对于 websocket , sse 使用上更为简便，只需要作用于 controller。

// home/home.controller.ts

import * as EventEmitter from 'events';
const myEmitter = new EventEmitter();

// --snip--

  @Sse('sse')
  sendServerMessage(): Observable<MessageEvent> {
    return new Observable<any>((observer) => {
      myEmitter.on('send', (data: number) => {
        observer.next({ data });
      });
    });
  }

  @Post('/addOnline')
  async addOnlineUser(@Body() dto: any) {
    const { userId } = dto;
    const res = await this.homeService.addOnlineUser(userId);
    const count = await this.homeService.getOnlineUsers();
    myEmitter.emit('send', count);
    return res;
  }

  @Post('/deleteOnline')
  async removeOnlineUser(@Body() dto: any) {
    const { userId } = dto;
    const res = await this.homeService.removeOnlineUser(userId);
    const count = await this.homeService.getOnlineUsers();
    myEmitter.emit('send', count);
    return res;
  }

前端中获取，仍然是 http 协议

function getHomeSse() {
    const eventSource = new EventSource("http://localhost:3001/home/sse");
    eventSource.onmessage = ({ data }) => {
        const onlineUsers = JSON.parse(data);
        onlineUsersCount.value = onlineUsers.count;
    };
}