socket.io和节点集群：将事件分派到套接字

Question

我有一个nodejs集群服务器，它使用mongo changestream侦听器，通过socket.io向客户端发送数据。

< code>{ userId: 'aaa '，socketId: 'bbb' }

用于存储此数据的redis客户端在主进程中初始化。mongo变更流是在主进程中创建的。

当变更流看到新文档时，它会将该文档作为消息发送到子流程。当子进程收到消息时，它可以从文档中检索 userId。使用 userId，可以从 redis 检索客户端连接的套接字 ID。

我遇到的问题是在从 redis 检索到套接字 Id 后尝试发出消息。我正在创建一个包含套接字 Id 的套接字处理程序对象。当我使用此套接字 Id 发出套接字消息时，如下所示：

io.sockets.to(userSocketId)
          .emit("confirmOrder", "Your order is being processed!")

我收到一个错误：

(node:31804)UnhandledPromiseRejectionWarning:错误:客户端在新的ClientClosedError处关闭(/Users/a 99999999/code/*/node _ modules/@ node-redis/client/dist/lib/errors . js:24:9)

错误来自redis，起源于上面写的套接字发射行。^^

下面是来自工作进程的更多代码：

const pubClient = createClient({ host: "127.0.0.1", port: 6379 }),
    subClient = pubClient.duplicate();
  io.adapter(createAdapter(pubClient, subClient));
  setupWorker(io);

  io.on("connection", (socket) => {
    const socketId = socket.id;

    socket.emit("connection", "SERVER: you are connected");

    socket.on("userConnect", (user) => {
      let { userId } = user;
      userConnectClient
        .HSET(userId, { userId, socketId })
        .catch((err) => console.log("ERROR: ", err));
    });
  });

  process.on("message", async ({ type, data }) => {
    switch (type) {
      case "dispatch:order":
        let { order } = JSON.parse(data);
        const socketsHandler = await createSocketsHandler(order);

        const userSocketId = socketsHandler.user.socketId;
        io.sockets
          .to(userSocketId)
          .emit("confirmOrder", "Your order is being processed!");
        break;
    }
  });

  async function createSocketsHandler(order) {
    let { userId } = order;
    let user = await userConnectClient
      .HGETALL(userId)
      .catch((err) => console.log(err));
    return {
      user: user,
    };
  }

在这一点上我暂时被难住了。目前正在试验io对象，并试图找到更好的工具来监控redis。感谢任何帮助/问题！谢谢大家！

Answer 1

从那时起，我就意识到redis客户端工作不正常的原因。我正在使用redis的发布者和订户客户端。问题是，我在服务器的工作进程中创建redis客户端。因此，每当服务器向redis发出命令时，它都无法正确执行，因为每个工作进程都有一对客户端，我认为这不是正确的实现。

这是通过在我的集群服务器代码之外创建redis客户端来解决的。； P我的服务器现在可以在主进程和工作进程中正确订阅redis客户端了！