JavaScript WebSocket 通信中怎么实现支持频道订阅的通信模型-js教程-PHP中文网

websocket需在应用层实现频道订阅模型，通过subscribe/unsubscribe控制消息、服务端map维护channel→set订阅关系、客户端封装事件监听，再扩展权限校验、redis集群支持与心跳重连。

javascript websocket 通信中怎么实现支持频道订阅的通信模型

WebSocket 本身不内置频道（channel）或主题（topic）概念，要实现“订阅/取消订阅频道”的通信模型，需在应用层设计协议约定和状态管理。核心思路是：客户端发送带频道标识的控制消息（如 SUBSCRIBE），服务端维护每个连接的订阅关系，并在广播时按频道过滤目标连接。

定义轻量级通信协议

双方约定简单 JSON 消息格式，区分控制指令与业务数据：

SUBSCRIBE：客户端请求加入频道，例如 {"type":"SUBSCRIBE","channel":"chat/general"}
UNSUBSCRIBE：退出频道，例如 {"type":"UNSUBSCRIBE","channel":"chat/general"}
PUBLISH：向某频道发消息，例如 {"type":"PUBLISH","channel":"chat/general","data":{"user":"Alice","text":"Hi!"}}
MESSAGE：服务端推送的频道消息（单播或广播），例如 {"type":"MESSAGE","channel":"chat/general","data":{...}}

服务端维护订阅关系（Node.js + ws 示例）

使用 Map 结构管理：`channel → Set` 存储所有订阅该频道的活跃连接；同时为每个 WebSocket 维护其订阅的频道集合（便于断连时自动清理）：

Java Maven Code Review

审查Java Maven项目（ZIP压缩包或GitLab仓库URL），检查代码规范、命名、模块边界、可维护性问题以及重复代码。

下载

const channels = new Map(); // 'chat/general' → Set<websocket>
const clientSubs = new WeakMap(); // ws → Set<string>

ws.on('message', (data) => {
  try {
    const msg = JSON.parse(data);
    if (msg.type === 'SUBSCRIBE' && msg.channel) {
      const channel = msg.channel;
      if (!channels.has(channel)) channels.set(channel, new Set());
      channels.get(channel).add(ws);

      let subs = clientSubs.get(ws);
      if (!subs) {
        subs = new Set();
        clientSubs.set(ws, subs);
      }
      subs.add(channel);
    } else if (msg.type === 'UNSUBSCRIBE' && msg.channel) {
      const channel = msg.channel;
      channels.get(channel)?.delete(ws);
      clientSubs.get(ws)?.delete(channel);
    } else if (msg.type === 'PUBLISH' && msg.channel && msg.data !== undefined) {
      const channel = msg.channel;
      const payload = { type: 'MESSAGE', channel, data: msg.data };
      const clients = channels.get(channel);
      if (clients) {
        clients.forEach(client => {
          if (client.readyState === WebSocket.OPEN) {
            client.send(JSON.stringify(payload));
          }
        });
      }
    }
  } catch (e) {
    ws.send(JSON.stringify({ type: 'ERROR', reason: 'Invalid message' }));
  }
});

// 断连时自动清理
ws.on('close', () => {
  const subs = clientSubs.get(ws);
  if (subs) {
    for (const channel of subs) {
      channels.get(channel)?.delete(ws);
    }
  }
});
</string></websocket>

客户端封装订阅逻辑（浏览器端）

避免每次手动拼 JSON，封装 subscribe()、unsubscribe() 和 publish() 方法：

class ChannelSocket {
  constructor(url) {
    this.ws = new WebSocket(url);
    this.ws.onmessage = (e) => {
      const msg = JSON.parse(e.data);
      if (msg.type === 'MESSAGE' && msg.channel && msg.data !== undefined) {
        const handlers = this.handlers[msg.channel] || [];
        handlers.forEach(cb => cb(msg.data));
      }
    };
  }

  subscribe(channel, callback) {
    this.ws.send(JSON.stringify({ type: 'SUBSCRIBE', channel }));
    if (!this.handlers) this.handlers = {};
    if (!this.handlers[channel]) this.handlers[channel] = [];
    this.handlers[channel].push(callback);
  }

  unsubscribe(channel, callback) {
    this.ws.send(JSON.stringify({ type: 'UNSUBSCRIBE', channel }));
    const handlers = this.handlers[channel];
    if (handlers) {
      const idx = handlers.indexOf(callback);
      if (idx > -1) handlers.splice(idx, 1);
    }
  }

  publish(channel, data) {
    this.ws.send(JSON.stringify({ type: 'PUBLISH', channel, data }));
  }
}

// 使用示例
const socket = new ChannelSocket('ws://localhost:3000');
socket.subscribe('chat/general', (data) => console.log('收到:', data));
socket.publish('chat/general', { user: 'Bob', text: 'Hello world' });