Node.js中Redis Pub/Sub结合WebSocket实现选择性消息推送问题求助

问题背景 #

我在Node.js中尝试结合Redis Pub/Sub与WebSocket实现消息订阅推送功能，核心需求是仅向订阅对应频道的WebSocket客户端推送消息。比如3个订阅者中，两个订阅BOARD-A频道，一个订阅BOARD-B频道，当向BOARD-A发布消息时，只有前两个订阅者能收到，但初始代码会推送给所有活跃订阅者。

初始代码的核心问题 #

const WebSocket = require('ws');
const Redis = require('ioredis');

// 创建WebSocket服务器
const wss = new WebSocket.Server({ port: 8080 });

// 创建Redis订阅客户端
const redisSubscriber = new Redis();

// 创建Redis发布客户端
const redisPublisher = new Redis();

// WebSocket连接建立事件
wss.on('connection', (ws) => {
  // 接收客户端消息事件
  ws.on('message', (message) => {
    const data = JSON.parse(message);

    if (data.event === 'subscribe') {
      // 订阅Redis频道
      redisSubscriber.subscribe(data.channel);
      // 绑定Redis消息接收回调
      redisSubscriber.on('message', (channel, message) => {
        // 向客户端发送Redis消息
        ws.send(JSON.stringify({ channel, message }));
      });
    } else if (data.event === 'publish') {
      // 向Redis频道发布消息
      redisPublisher.publish(data.channel, data.message);
    }
  });

  // WebSocket连接关闭事件
  ws.on('close', () => {
    // 取消所有Redis频道订阅
    redisSubscriber.unsubscribe();
  });
});

• 重复绑定消息回调：每个WebSocket客户端订阅频道时，都会给同一个redisSubscriber实例绑定一次message事件。Redis收到消息时，所有绑定过的回调都会执行，导致所有订阅过任何频道的客户端都能收到消息。
• 全局取消订阅：单个客户端断开时，调用redisSubscriber.unsubscribe()会取消所有频道的订阅，直接影响其他正常订阅的客户端。
• 无客户端-频道映射：没有维护每个频道对应的WebSocket客户端集合，无法实现精准推送。

更新后代码的修复与优化 #

const WebSocket = require('ws');
const Redis = require('ioredis');

const wss = new WebSocket.Server({ port: 8080 });

const subscriber = new Redis();
const publisher = new Redis();

// 维护频道到WebSocket客户端的映射：频道名 -> 客户端集合
const channelClients = new Map();

wss.on('connection', (ws) => {
  ws.on('message', (msg) => {
    const data = JSON.parse(msg);

    if (data.event === 'subscribe') {
      // 频道不存在时，初始化集合并订阅Redis频道
      if (!channelClients.has(data.channel)) {
        channelClients.set(data.channel, new Set());
        subscriber.subscribe(data.channel);
      }
      // 将当前客户端加入对应频道的集合
      channelClients.get(data.channel).add(ws);
    }
    else if (data.event === 'publish') {
      // 向Redis频道发布消息
      publisher.publish(data.channel, data.message);
    }
  });

  ws.on('close', () => {
    // 遍历所有频道，移除当前客户端
    for (const [channel, clients] of channelClients.entries()) {
      clients.delete(ws);
      // 频道无订阅者时，取消Redis订阅并删除映射
      if (clients.size === 0) {
        subscriber.unsubscribe(channel);
        channelClients.delete(channel);
      }
    }
  });
});

// 全局绑定Redis消息接收回调，仅推送对应频道的客户端
subscriber.on('message', (channel, message) => {
  const clients = channelClients.get(channel);
  if (clients) {
    for (const client of clients) {
      // 检查WebSocket连接状态，避免发送失败
      if (client.readyState === WebSocket.OPEN) {
        client.send(JSON.stringify({ channel, message }));
      }
    }
  }
});

关键修复点 #

• 频道-客户端映射：用channelClients存储每个频道对应的WebSocket客户端集合，实现精准推送。
• 单一全局回调：仅给subscriber绑定一次message事件，收到消息后根据映射找到对应客户端推送，避免重复回调。
• 精准取消订阅：客户端断开时，仅从对应频道的集合中移除自身；当频道无订阅者时，才取消Redis订阅，不影响其他客户端。
• 连接状态校验：推送前验证WebSocket连接是否处于OPEN状态，避免发送失败报错。

可进一步优化的方向 #

• JSON解析异常处理：客户端发送的消息可能不是合法JSON，需用try/catch包裹JSON.parse(msg)，避免服务崩溃。
• 多频道订阅支持：当前逻辑默认一个客户端订阅一个频道，可修改为支持同一客户端订阅多个频道（比如在ws实例上挂载已订阅频道列表，断开时遍历移除）。
• 订阅去重提示：同一客户端重复订阅同一频道时，可添加提示逻辑（虽然Set会自动去重，但可以告知客户端无需重复订阅）。
• 异常监听：给Redis客户端添加error事件监听，处理连接失败等异常；WebSocket发送失败时捕获error事件。
• 心跳检测：添加WebSocket心跳机制，清理僵尸连接，避免无效客户端占用资源。

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