PM2 多进程部署 Socket.IO，如何配置粘性会话？-字节棱镜 (Byte Prism)

PM2 Cluster 下 Socket.IO 偶发断连，仅调整 Worker 数不能解决会话归属和跨 Worker 消息分发问题；需要让同一 Engine.IO 会话固定在一个 Worker，并用适配器把消息送到其他 Worker。

掘金热榜文章《多进程环境下使用 PM2 部署 Socket.IO：需要注意什么》讨论了这个问题。排查时应区分连接归属、消息分发和业务状态：粘性会话负责连接归属，Redis 适配器负责跨 Worker 广播，登录态、计数器等状态则必须外置。如果不区分这三层，排查时容易反复调整 Worker 数，却找不到根因。

先画清连接经过哪些进程

典型链路可简化为：

浏览器或 SDK
    ↓
CDN / 负载均衡
    ↓
Nginx
    ↓
PM2 主进程或共享监听入口
    ↓
Worker A
    ↓
Engine.IO：握手、会话、传输升级、心跳
    ↓
Socket.IO：namespace、room、事件

PM2 Cluster 的多个 Worker 共享监听入口，请求进入后再由进程调度器分配给具体 Worker。Node.js Cluster 的典型调度方式也由主进程完成连接分发；连接建立后的数据不再逐次经过主进程。但 Socket.IO 并不是原生 WebSocket，Engine.IO 位于其下层，负责握手、升级、心跳和传输方式，因此一次 Socket.IO 连接通常不只是一条 HTTP 请求。

默认 polling 模式下，客户端可能先经历以下步骤：

发起一次不带 sid 的 polling 握手。
服务端生成 Engine.IO 会话 ID。
继续通过 polling 收发事件。
尝试把连接升级为 WebSocket。
在后续连接中维持心跳。

这些步骤可能使用不同 HTTP 连接。在单进程部署中，它们都落在同一进程；在 PM2 Cluster 中，polling 握手和 WebSocket 升级可能分别进入 Worker A 与 Worker B。Worker B 看不到 Worker A 创建的会话，于是出现 Session ID unknown、升级失败或频繁退回 polling。

因此，“浏览器显示已连接，但消息偶尔丢失”通常与跨 Worker 缺少共享消息链路有关：连接本身可能暂时成立，后续事件却无法送达其他 Worker 上的客户端。

搭出可复现的最小工程

下面以 Node.js 22、Socket.IO 4.8.1 的 API 写法为基准。正式部署前应在目标 Node.js、PM2 和 Redis 版本上运行验收脚本，并提交 package-lock.json，让生产构建使用 npm ci 固定依赖。

目录可以先保持简单：

socketio-pm2/
├── package.json
├── server.js
├── ecosystem.config.cjs
├── scripts/
│   └── verify.js
└── logs/

package.json：

{
  "name": "socketio-pm2-guide",
  "version": "1.0.0",
  "private": true,
  "engines": {
    "node": ">=20"
  },
  "scripts": {
    "start": "node server.js",
    "start:pm2": "pm2 start ecosystem.config.cjs --env production",
    "verify": "node scripts/verify.js"
  },
  "dependencies": {
    "@socket.io/redis-adapter": "^8.3.0",
    "pm2": "^5.4.3",
    "redis": "^4.7.0",
    "socket.io": "^4.8.1",
    "socket.io-client": "^4.8.1"
  }
}

server.js 同时提供健康检查、Engine.IO 会话日志、房间消息、定向消息和 Redis 适配器：

"use strict";

const http = require("node:http");
const { Server } = require("socket.io");
const { createClient } = require("redis");
const { createAdapter } = require("@socket.io/redis-adapter");

const PORT = Number(process.env.PORT || 3000);
const HOST = process.env.HOST || "0.0.0.0";
const WORKER_ID =
  process.env.APP_INSTANCE || process.env.NODE_APP_INSTANCE || "standalone";
const REDIS_KEY = process.env.REDIS_KEY || "socketio";
const FORCE_EXIT_MS = Number(process.env.FORCE_EXIT_MS || 10000);

const redisClients = [];
let shuttingDown = false;

function log(event, fields = {}) {
  console.log(
    JSON.stringify({
      time: new Date().toISOString(),
      event,
      pid: process.pid,
      worker: WORKER_ID,
      ...fields
    })
  );
}

function createHttpServer() {
  const server = http.createServer((req, res) => {
    const pathname = new URL(req.url, "http://localhost").pathname;

    if (pathname === "/healthz") {
      res.writeHead(200, { "content-type": "application/json" });
      res.end(
        JSON.stringify({
          ok: true,
          pid: process.pid,
          worker: WORKER_ID,
          uptime: process.uptime()
        })
      );
      return;
    }

    res.writeHead(404, { "content-type": "application/json" });
    res.end(JSON.stringify({ error: "not_found" }));
  });

  server.on("request", (req) => {
    if (!req.url || !req.url.startsWith("/socket.io/")) {
      return;
    }

    const params = new URL(req.url, "http://localhost").searchParams;
    const sid = params.get("sid");
    const transport = params.get("transport");

    if (!sid && transport !== "polling" && !req.headers.upgrade) {
      return;
    }

    log("engine.request", {
      sid,
      transport,
      upgrade: req.headers.upgrade || null
    });
  });

  server.on("upgrade", (req) => {
    const params = new URL(req.url || "/", "http://localhost").searchParams;

    log("engine.upgrade.request", {
      sid: params.get("sid"),
      transport: params.get("transport"),
      upgrade: req.headers.upgrade || null
    });
  });

  return server;
}

async function attachRedisAdapter(io) {
  if (!process.env.REDIS_URL) {
    throw new Error("REDIS_URL is required");
  }

  const pubClient = createClient({ url: process.env.REDIS_URL });
  const subClient = pubClient.duplicate();

  for (const [name, client] of [
    ["pub", pubClient],
    ["sub", subClient]
  ]) {
    client.on("error", (error) => {
      log("redis.error", { client: name, message: error.message });
    });
  }

  await Promise.all([pubClient.connect(), subClient.connect()]);
  redisClients.push(pubClient, subClient);

  io.adapter(createAdapter(pubClient, subClient, { key: REDIS_KEY }));
  log("redis.adapter.ready", { key: REDIS_KEY });
}

async function shutdown(io, signal) {
  if (shuttingDown) {
    return;
  }

  shuttingDown = true;
  log("shutdown.start", { signal });

  const forceTimer = setTimeout(() => {
    log("shutdown.force", { timeout: FORCE_EXIT_MS });
    process.exit(1);
  }, FORCE_EXIT_MS);
  forceTimer.unref();

  try {
    await new Promise((resolve) => io.close(() => resolve()));
    await Promise.allSettled(
      redisClients.map((client) =>
        client.isOpen ? client.quit() : Promise.resolve()
      )
    );

    clearTimeout(forceTimer);
    log("shutdown.done", { signal });
    process.exit(0);
  } catch (error) {
    log("shutdown.error", { message: error.message });
    process.exit(1);
  }
}

async function main() {
  const httpServer = createHttpServer();

  const io = new Server(httpServer, {
    path: "/socket.io/",
    serveClient: false,
    cookie: {
      name: process.env.IO_COOKIE_NAME || "io",
      path: "/",
      httpOnly: true,
      sameSite: "lax",
      secure: process.env.COOKIE_SECURE === "true"
    }
  });

  await attachRedisAdapter(io);

  io.engine.on("connection", (engineSocket) => {
    log("engine.connection", {
      engineId: engineSocket.id,
      transport: engineSocket.transport.name,
      address: engineSocket.request?.socket?.remoteAddress || null
    });
  });

  io.on("connection", (socket) => {
    const session = {
      engineId: socket.conn.id,
      socketId: socket.id,
      pid: process.pid,
      worker: WORKER_ID,
      transport: socket.conn.transport.name,
      address: socket.handshake.address
    };

    log("socket.connection", session);
    socket.emit("session:info", session);

    socket.conn.on("upgrade", (transport) => {
      const upgradeInfo = {
        ...session,
        transport: transport?.name || "unknown"
      };

      log("engine.upgrade", upgradeInfo);
      socket.emit("session:upgrade", upgradeInfo);
    });

    async function joinRoom(room, ack) {
      if (typeof room !== "string" || room.length === 0) {
        if (typeof ack === "function") {
          ack({ ok: false });
        }
        return;
      }

      const normalizedRoom = room.slice(0, 100);
      await socket.join(normalizedRoom);

      if (typeof ack === "function") {
        ack({
          ok: true,
          pid: process.pid,
          worker: WORKER_ID
        });
      }
    }

    socket.on("room:join", (payload, ack) => {
      void joinRoom(payload?.room, ack);
    });

    socket.on("verify:join", (payload, ack) => {
      void joinRoom(payload?.room, ack);
    });

    socket.on("broadcast", (payload) => {
      io.emit("broadcast", {
        payload,
        from: socket.id,
        pid: process.pid
      });
    });

    socket.on("room:message", (payload) => {
      if (typeof payload?.room !== "string") {
        return;
      }

      socket.to(payload.room).emit("room:message", {
        message: payload.message,
        from: socket.id,
        pid: process.pid
      });
    });

    socket.on("direct:message", (payload) => {
      if (typeof payload?.target !== "string") {
        return;
      }

      io.to(payload.target).emit("direct:message", {
        message: payload.message,
        from: socket.id,
        pid: process.pid
      });
    });

    socket.on("verify:trigger", (payload) => {
      if (typeof payload?.room !== "string") {
        return;
      }

      io.to(payload.room).emit("verify:broadcast", {
        messageId: payload.messageId,
        pid: process.pid,
        worker: WORKER_ID
      });
    });

    socket.on("verify:global", (payload) => {
      io.emit("verify:global", {
        messageId: payload.messageId,
        pid: process.pid,
        worker: WORKER_ID
      });
    });

    socket.on("verify:direct", (payload) => {
      if (typeof payload?.target !== "string") {
        return;
      }

      io.to(payload.target).emit("verify:direct", {
        messageId: payload.messageId,
        pid: process.pid,
        worker: WORKER_ID
      });
    });

    socket.on("disconnect", (reason) => {
      log("socket.disconnect", {
        engineId: socket.conn.id,
        socketId: socket.id,
        reason
      });
    });
  });

  process.once("SIGTERM", () => {
    void shutdown(io, "SIGTERM");
  });

  process.once("SIGINT", () => {
    void shutdown(io, "SIGINT");
  });

  httpServer.once("error", (error) => {
    log("server.error", { message: error.message });
    process.exit(1);
  });

  httpServer.listen(PORT, HOST, () => {
    log("server.listening", {
      host: HOST,
      port: PORT,
      redisKey: REDIS_KEY
    });
  });
}

main().catch((error) => {
  log("startup.error", { message: error.message });
  process.exit(1);
});

ecosystem.config.cjs：

module.exports = {
  apps: [
    {
      name: "socketio-pm2",
      script: "server.js",
      cwd: __dirname,
      interpreter: "node",
      instances: Number(process.env.WORKERS || 4),
      exec_mode: "cluster",
      watch: false,
      autorestart: true,
      listen_timeout: 10000,
      kill_timeout: 15000,
      max_memory_restart: process.env.MAX_MEMORY || "512M",
      out_file: "./logs/pm2-out.log",
      error_file: "./logs/pm2-error.log",
      merge_logs: false,
      time: true,
      env: {
        NODE_ENV: "production",
        HOST: "0.0.0.0",
        PORT: "3000",
        REDIS_URL: "redis://127.0.0.1:6379",
        REDIS_KEY: "socketio",
        IO_COOKIE_NAME: "io",
        COOKIE_SECURE: "false",
        FORCE_EXIT_MS: "10000"
      }
    }
  ]
};

512M 只是示例保护值，不应代替内存泄漏排查。PM2 可以按内存排序进程并使用 pm2 monit 观察变化，但 max_memory_restart 更适合作为失控后的止损线，而不是泄漏修复方案。

固定 instances: 4 便于复现和验收。生产环境可以改成 instances: "max" 让 PM2 按可用 CPU 资源扩展，但 WebSocket 还会占用连接对象、缓冲区、订阅和业务状态，Worker 数不能只按 CPU 核数决定；Node.js 提供的 availableParallelism() 也只是并行度参考，而不是内存容量参考。

启动和查看进程：

npm install
mkdir -p logs
pm2 start ecosystem.config.cjs --env production
pm2 list
pm2 logs socketio-pm2 --lines 100 --nostream
pm2 list --sort memory
pm2 monit socketio-pm2

pm2 startup 会输出与当前平台匹配的初始化命令，执行该命令后再保存进程清单：

pm2 startup
pm2 save

应用日志已经统一带上 pid 和 worker。排查时不要只看“进程在线”，还要确认同一个 Engine.IO sid 究竟落在哪个 PID 上。

让握手与升级落在同一 Worker

一个典型失败过程如下：

polling 握手 → Worker A
生成 sid = abc123

WebSocket 升级，携带 sid=abc123 → Worker B
Worker B 查不到 abc123
→ Session ID unknown

PM2 Cluster 与 Node.js Cluster 都涉及主进程或调度器向 Worker 分发连接，底层模式包括分发、连接移交和进程替换。判断重点是 Engine.IO 会话是否被分配给同一 Worker，而不是“有没有主进程”。

不同代理拓扑能提供的粘性层级并不相同：

部署拓扑	实际可固定的层级	主要限制
客户端直连 PM2 Cluster	Worker	需要 Worker 级会话保持机制，或只使用 WebSocket
Nginx 后接一个 PM2 Cluster	Worker	Nginx 只看到一个共享入口，普通 `ip_hash` 不能选择内部 Worker
Nginx 后接多个单进程 PM2 实例	PM2 实例	可以按 Cookie 或源 IP 做上游粘性
外部负载均衡后接多个 PM2 Cluster	外部实例及内部 Worker	外层粘性不能自动替代内层 Worker 粘性
多个单进程容器	容器或 Pod	由负载均衡器提供 Cookie、源 IP 或平台级会话保持

对单个 PM2 Cluster，直接配置下面这段通常并不能解决问题：

upstream socket_io_backend {
    ip_hash;
    server 127.0.0.1:3000;
}

因为 Nginx 的上游只有 127.0.0.1:3000 这一个入口。ip_hash 只能决定请求进入哪个上游入口，无法决定 PM2 再把请求交给哪个内部 Worker。

如果业务必须保留 polling 兼容能力，优先选择能够按进程路由的拓扑。一种替代拓扑是让 PM2 启动多个单进程实例，每个实例监听独立端口，再由 Nginx 按 io Cookie 做粘性：

// ecosystem.instances.config.cjs
const ports = [3001, 3002, 3003, 3004];

module.exports = {
  apps: ports.map((port, index) => ({
    name: `socketio-${port}`,
    script: "server.js",
    cwd: __dirname,
    interpreter: "node",
    exec_mode: "fork",
    watch: false,
    autorestart: true,
    kill_timeout: 15000,
    max_memory_restart: "512M",
    env: {
      NODE_ENV: "production",
      HOST: "0.0.0.0",
      PORT: String(port),
      APP_INSTANCE: String(index),
      REDIS_URL: "redis://127.0.0.1:6379",
      REDIS_KEY: "socketio",
      IO_COOKIE_NAME: "io"
    }
  }))
};

对应的 Nginx 上游可以是：

upstream socket_io_pool {
    hash $cookie_io consistent;

    server 127.0.0.1:3001;
    server 127.0.0.1:3002;
    server 127.0.0.1:3003;
    server 127.0.0.1:3004;
}

这不再是 PM2 原生 Cluster 的共享端口模式，而是“PM2 管理多个独立进程，Nginx 负责进程级粘性”。不要同时启动两套配置。浏览器可以自动保存并回传 io Cookie；Node.js 客户端是否维护 Cookie 取决于客户端实现和传输方式，不能默认浏览器行为一定存在。

源 IP 粘性也可以使用 ip_hash，但同一办公网、运营商 NAT 或移动网络出口可能对应大量用户，把它们集中到同一个实例。

可评估 @socket.io/sticky 是否能提供 Worker 级粘性：

npm install @socket.io/sticky

安装依赖不等于粘性已经生效。该方案与 Node.js Cluster 的连接调度机制有关，而 PM2 有自己的 Cluster 进程管理和调度路径。应先核对当前 @socket.io/sticky、PM2、Node.js 和 Socket.IO 版本的官方说明及兼容范围，再按其当前用法接入。只有下面的日志能证明同一 sid 始终对应同一 PID，才算配置完成：

engine.connection          sid=abc123 pid=101
engine.request             sid=abc123 pid=101
engine.upgrade.request     sid=abc123 pid=101
socket.connection          engineId=abc123 pid=101

如果业务可以接受只使用 WebSocket，优先采用下面的客户端配置。它没有 polling 握手和后续升级两个阶段，连接建立后由同一条 TCP 连接承载，链路更短，也更容易判断故障位于 PM2、代理还是客户端：

const socket = io({
  path: "/socket.io/",
  transports: ["websocket"],
  upgrade: false,
  reconnection: true
});

代价是失去 polling 回退能力，因此这应被视为明确的产品能力和网络兼容性取舍，不能作为无条件的“最佳方案”。

补齐跨 Worker 消息链路

粘性会话只回答“连接归谁处理”，不回答“消息如何送达其他 Worker”。

上面的服务已经使用两个独立 Redis 客户端接入适配器：

const pubClient = createClient({ url: process.env.REDIS_URL });
const subClient = pubClient.duplicate();

await Promise.all([pubClient.connect(), subClient.connect()]);

io.adapter(createAdapter(pubClient, subClient, {
  key: process.env.REDIS_KEY || "socketio"
}));

所有 Worker 必须使用相同的 Redis 地址、认证信息和 key。如果某个 Worker 的 REDIS_KEY 不同，它会加入另一组频道，看起来就像“部分客户端收不到广播”。

三种发送方式的语义不同：

// 广播给所有 Worker 上的所有客户端
io.emit("broadcast", payload);

// 广播给某个房间，依赖跨 Worker 的 room 成员信息
io.to("orders").emit("room:message", payload);

// 发送给一个 Socket.IO socket ID
io.to(targetSocketId).emit("direct:message", payload);

io.to(socketId) 接收的是 Socket.IO ID，不是 Engine.IO 的 sid。在生产代码中应使用 socket.id，不要把 Engine.IO sid 当成定向发送目标。

三种机制不能互相替代：

机制	解决的问题	不解决的问题
粘性会话	握手、轮询和升级由同一 Worker 处理	跨 Worker 广播、进程内状态同步
Redis 适配器	跨 Worker 广播、房间和定向事件	连接归属、登录态和计数同步
外部共享状态	登录态、计数器、缓存、幂等记录	路由和消息分发

socket.data 只属于当前 Socket 对象。不同 Worker 上的 socket.data、本地会话表、内存计数器和定时任务不会自动同步。多进程下，不能再把进程内存当作全局状态，这类状态应放进 Redis 或其他共享存储。

排查“部分客户端收不到消息”时，还要区分两种情况：

io.emit() 根本没有到达其他 Worker 上的客户端，通常是适配器、Redis 地址或频道前缀不一致。
全局广播已经到达其他 Worker，但某个客户端仍未收到房间消息，可能是该客户端重连后没有重新加入房间。

后者不是广播链路故障，而是客户端生命周期问题。

让代理正确转发 WebSocket

Nginx 的 map 必须放在 http 上下文：

http {
    map $http_upgrade $connection_upgrade {
        default upgrade;
        ''      close;
    }

    upstream socket_io_backend {
        server 127.0.0.1:3000;
        keepalive 16;
    }

    server {
        listen 80;
        server_name _;

        location /socket.io/ {
            proxy_pass http://socket_io_backend;
            proxy_http_version 1.1;

            proxy_set_header Upgrade $http_upgrade;
            proxy_set_header Connection $connection_upgrade;
            proxy_set_header Host $host;

            proxy_set_header X-Real-IP $remote_addr;
            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
            proxy_set_header X-Forwarded-Proto $scheme;

            proxy_read_timeout 3600s;
            proxy_send_timeout 3600s;
            proxy_buffering off;
        }

        location = /healthz {
            proxy_pass http://socket_io_backend;
        }
    }
}

这里有五个关键点：

/socket.io/ 必须与服务端 path 和客户端路径一致。
proxy_http_version 1.1 与 Upgrade、Connection 请求头需要配套。
polling 请求没有 Upgrade 头时，映射为 Connection: close，不要把所有请求都伪装成 WebSocket 升级。
X-Forwarded-For 和 X-Real-IP 用于保留客户端地址；CDN 场景下还要确认可信代理链，不能无条件信任任意来源的请求头。
读取超时应大于心跳间隔并留出余量。超时过短时，现象通常不是立即报错，而是连接被代理周期性关闭后不断重连。

健康检查应使用 /healthz，不要用 /socket.io/ 充当探针。Socket.IO 路径承载的是长连接和协议升级，不适合反复建立、销毁探测连接。

排查时先做两组对照：

SOCKET_URL=http://127.0.0.1:3000 npm run verify
SOCKET_URL=http://127.0.0.1:8080 npm run verify

第一组绕过 Nginx，第二组经过代理。如果直连稳定而代理不稳定，应先修正代理头、路径和超时；如果两组都出现相同的 sid 与 PID 漂移，则问题位于 PM2 或会话保持方案。

在容器中明确多进程管理方式

容器化时首先要确定由谁管理多进程：

模型	进程结构	会话保持位置	主要代价
单容器运行 PM2 Cluster	一个容器内多个 Worker	容器入口加 Worker 内部	容器内同时存在两层进程管理
单容器单进程，多容器扩容	每个容器一个 Node.js 进程	容器或 Pod 层	需要编排系统管理副本和会话保持

已有 Next.js 项目在 Docker 中运行 PM2 Cluster 的实践，说明这种进程结构可以用于容器部署。但它不能直接证明 Socket.IO 的 polling 粘性、跨 Worker 广播和容器停止信号已经正确，Socket.IO 仍需单独验收。

下面的 Dockerfile 可以作为单容器 PM2 Cluster 的起点：

FROM node:22-bookworm-slim

WORKDIR /app

COPY package*.json ./
RUN npm ci --omit=dev

COPY . .
RUN mkdir -p logs

ENV NODE_ENV=production
ENV HOST=0.0.0.0
ENV PORT=3000

EXPOSE 3000
STOPSIGNAL SIGTERM

HEALTHCHECK --interval=30s --timeout=3s --start-period=20s --retries=3 \
  CMD node -e "fetch('http://127.0.0.1:3000/healthz').then(r=>process.exit(r.ok?0:1)).catch(()=>process.exit(1))"

CMD ["./node_modules/.bin/pm2-runtime", "start", "ecosystem.config.cjs", "--env", "production"]

容器内必须监听 0.0.0.0，不能只监听 localhost。同时要核对端口发布、Redis 容器地址和健康检查目标。单个 PM2 Cluster 的 /healthz 请求可能落到任意 Worker，它只能证明共享入口当前有 Worker 可以响应，不能证明所有 Worker 都健康。

停止容器时，Docker 会向 PID 1 发送信号。pm2-runtime、PM2 和应用自身需要形成清晰的信号链：PM2 收到终止信号后给 Worker 留出 kill_timeout，应用收到 SIGTERM 后关闭 Socket.IO、HTTP 服务和 Redis 客户端。必须用实际的 docker stop 观察日志，确认进程没有被直接杀死，也没有跳过清理步骤。

在 Kubernetes 中，如果每个 Pod 已经运行 PM2 Cluster，再让 Deployment 创建多个副本，就会同时存在 Pod 级和 Worker 级两层调度。更简单的选择通常是每个 Pod 跑一个 Node.js 进程，由 Deployment 管理副本，再由 Ingress 或 Service 提供 Pod 级会话保持。PM2 Cluster 与 Kubernetes 都在处理进程分发、优雅退出和滚动替换，两层能力不应在没有必要时叠加。

把 Worker 重启做成可恢复流程

发布流程能逐步切换 HTTP 请求，但不会迁移既有长连接。Worker 被替换时，已经建立的 WebSocket 长连接必须重新建立。

服务端的 SIGTERM 处理应完成三件事：

停止接收新连接。
关闭 Socket.IO 客户端连接并停止 HTTP 服务。
等待在途任务、消息投递和 Redis 操作结束。

如果业务还有任务队列或数据库事务，应把对应队列标记为停止消费，并等待任务完成，而不是在 SIGTERM 回调里立即 process.exit()。PM2 的 kill_timeout 要大于应用自身的清理上限，留出退出余量。

pm2 reload socketio-pm2 --update-env 可以滚动替换 Cluster 中的 Worker，但被替换 Worker 上的现有长连接仍会断开。客户端需要带退避的重连策略：

const socket = io({
  path: "/socket.io/",
  reconnection: true,
  reconnectionDelay: 500,
  reconnectionDelayMax: 5000,
  timeout: 10000
});

socket.on("connect", () => {
  socket.emit(
    "session:resume",
    { token: currentToken },
    (result) => {
      if (!result?.ok) {
        return;
      }

      socket.emit("room:join", { room: "orders" }, () => {});
      socket.emit("subscription:resume", { topics: ["orders"] }, () => {});
    }
  );
});

重连后的恢复顺序建议固定为：

建立连接
  → 重新鉴权
  → 恢复业务订阅
  → 重新加入房间
  → 恢复消息处理

不要假设 socket.data、房间成员关系和服务端订阅会自动从旧连接迁移到新连接。

监控上应区分三类重启：

正常滚动发布：操作触发，Worker 逐个替换。
Worker 崩溃：进程异常退出，PM2 拉起新进程。
max_memory_restart 触发：内存超过配置阈值后被 PM2 主动重启。

三者都会导致长连接重建，但原因和恢复策略不同，日志中应保留触发来源。

用三项测试验收部署

“浏览器显示已连接”只能证明某个时刻存在连接，不能证明会话归属正确，也不能证明消息跨越了 Worker。

下面的验证脚本会记录连接、断开、错误、重连、Engine.IO 会话和接收结果，并测试全局广播、房间广播、定向消息以及指定 Worker 重启后的恢复。

scripts/verify.js：

"use strict";

const { io } = require("socket.io-client");

const url = process.env.SOCKET_URL || "http://127.0.0.1:3000";
const clientCount = Math.max(2, Number(process.env.CLIENTS || 6));
const room = process.env.ROOM || `verify-${Date.now()}`;
const transports = (process.env.TRANSPORTS || "polling,websocket")
  .split(",")
  .map((value) => value.trim())
  .filter(Boolean);

const clients = [];

function delay(ms) {
  return new Promise((resolve) => setTimeout(resolve, ms));
}

async function waitFor(predicate, timeoutMs, label) {
  const deadline = Date.now() + timeoutMs;

  while (Date.now() < deadline) {
    if (predicate()) {
      return;
    }
    await delay(100);
  }

  throw new Error(`Timeout: ${label}`);
}

function createClient(index) {
  const socket = io(url, {
    path: "/socket.io/",
    transports,
    reconnection: true,
    reconnectionDelay: 500,
    reconnectionDelayMax: 3000,
    timeout: 5000
  });

  const state = {
    index,
    socket,
    connected: false,
    joined: false,
    connectCount: 0,
    reconnectAttempts: [],
    session: null,
    upgrade: null,
    sessionChanged: false,
    roomMessages: new Set(),
    globalMessages: new Set(),
    directMessages: new Set(),
    disconnects: [],
    errors: []
  };

  socket.on("connect", () => {
    state.connected = true;
    state.connectCount += 1;

    socket.emit("verify:join", { room }, (reply) => {
      state.joined = Boolean(reply?.ok);
    });
  });

  socket.on("session:info", (info) => {
    if (
      state.session &&
      state.session.socketId !== info.socketId
    ) {
      state.sessionChanged = true;
    }
    state.session = info;
  });

  socket.on("session:upgrade", (info) => {
    state.upgrade = info;
  });

  socket.on("verify:broadcast", (message) => {
    if (message?.messageId) {
      state.roomMessages.add(message.messageId);
    }
  });

  socket.on("verify:global", (message) => {
    if (message?.messageId) {
      state.globalMessages.add(message.messageId);
    }
  });

  socket.on("verify:direct", (message) => {
    if (message?.messageId) {
      state.directMessages.add(message.messageId);
    }
  });

  socket.on("disconnect", (reason) => {
    state.connected = false;
    state.joined = false;
    state.disconnects.push(reason);
  });

  socket.on("connect_error", (error) => {
    state.errors.push(error.message);
  });

  socket.io.on("reconnect_attempt", (attempt) => {
    state.reconnectAttempts.push(attempt);
  });

  clients.push(state);
  return state;
}

function snapshot() {
  return clients.map((client) => ({
    index: client.index,
    connected: client.connected,
    joined: client.joined,
    connectCount: client.connectCount,
    session: client.session,
    upgrade: client.upgrade,
    roomMessages: [...client.roomMessages],
    globalMessages: [...client.globalMessages],
    directMessages: [...client.directMessages],
    disconnects: client.disconnects,
    errors: client.errors
  }));
}

function closeAll() {
  for (const client of clients) {
    client.socket.close();
  }
}

async function run() {
  for (let index = 0; index < clientCount; index += 1) {
    createClient(index);
  }

  await waitFor(
    () => clients.every((client) => client.connected && client.session),
    15000,
    "clients connected"
  );

  await waitFor(
    () => clients.every((client) => client.joined),
    10000,
    "clients joined room"
  );

  await waitFor(
    () =>
      clients.every(
        (client) =>
          client.session?.transport === "websocket" ||
          client.upgrade?.transport === "websocket"
      ),
    10000,
    "websocket upgrade"
  );

  const sessionMismatches = clients.filter(
    (client) =>
      client.upgrade && client.session.pid !== client.upgrade.pid
  );

  const workerPids = new Set(
    clients.map((client) => client.session?.pid).filter(Boolean)
  );

  if (workerPids.size < 2) {
    throw new Error(
      "Only one Worker was observed; cross-Worker delivery was not proven"
    );
  }

  if (sessionMismatches.length > 0) {
    throw new Error("Session and upgrade were handled by different Workers");
  }

  await delay(500);

  const roomMessageId = `room-${Date.now()}`;
  clients[0].socket.emit("verify:trigger", {
    room,
    messageId: roomMessageId
  });

  await waitFor(
    () =>
      clients.every((client) =>
        client.roomMessages.has(roomMessageId)
      ),
    10000,
    "cross-Worker room broadcast"
  );

  const globalMessageId = `global-${Date.now()}`;
  clients[0].socket.emit("verify:global", {
    messageId: globalMessageId
  });

  await waitFor(
    () =>
      clients.every((client) =>
        client.globalMessages.has(globalMessageId)
      ),
    10000,
    "global broadcast"
  );

  const directTarget = clients[1];
  const directMessageId = `direct-${Date.now()}`;

  clients[0].socket.emit("verify:direct", {
    target: directTarget.session.socketId,
    messageId: directMessageId
  });

  await waitFor(
    () => directTarget.directMessages.has(directMessageId),
    10000,
    "direct message"
  );

  let restartResult = {
    requested: false,
    affectedClients: 0
  };

  if (process.env.RESTART_PID) {
    const targetPid = Number(process.env.RESTART_PID);
    const affected = clients.filter(
      (client) => client.session?.pid === targetPid
    );

    if (affected.length === 0) {
      throw new Error(
        "RESTART_PID was not connected to any verification client"
      );
    }

    const previousConnectCount = new Map(
      affected.map((client) => [client, client.connectCount])
    );

    process.kill(targetPid, "SIGTERM");

    await waitFor(
      () =>
        affected.every(
          (client) =>
            client.connectCount >
              previousConnectCount.get(client) &&
            client.sessionChanged &&
            client.joined
        ),
      20000,
      "clients reconnect after Worker restart"
    );

    await delay(500);

    const afterRestartMessageId = `restart-${Date.now()}`;
    clients[0].socket.emit("verify:trigger", {
      room,
      messageId: afterRestartMessageId
    });

    await waitFor(
      () =>
        clients.every((client) =>
          client.roomMessages.has(afterRestartMessageId)
        ),
      10000,
      "broadcast after Worker restart"
    );

    restartResult = {
      requested: true,
      affectedClients: affected.length,
      reconnected: true
    };
  }

  const result = {
    ok: true,
    url,
    transports,
    clientCount,
    workerPids: [...workerPids],
    sessionMismatches: sessionMismatches.length,
    roomBroadcast: true,
    globalBroadcast: true,
    directMessage: true,
    restart: restartResult
  };

  console.log(JSON.stringify(result, null, 2));
}

run()
  .catch((error) => {
    process.exitCode = 1;
    console.error(
      JSON.stringify(
        {
          ok: false,
          error: error.message,
          clients: snapshot()
        },
        null,
        2
      )
    );
  })
  .finally(() => {
    closeAll();
  });

先运行默认的 polling 升级测试：

npm run verify

再把 WebSocket-only 作为对照：

TRANSPORTS=websocket npm run verify

从 pm2 list 找到一个实际承载测试连接的 Worker PID，再执行：

RESTART_PID=12345 npm run verify

验收结果至少应满足：

测试	通过标准
会话归属	同一 `engineId` 的握手、轮询和升级始终由同一 PID 处理
跨 Worker 广播	至少观察到两个 PID，全局、房间和定向消息全部到达
故障恢复	指定 Worker 重启后，受影响客户端自动重连、重新入房并继续收消息

同时用下面的命令观察进程和内存：

pm2 list --sort memory
pm2 monit socketio-pm2

PM2 的内存监控和进程列表适合发现持续增长与异常重启，但验收仍应回到具体 sid、PID 和消息送达结果上。

按现象倒查具体故障

现象	优先怀疑	检查与处理
握手成功，但升级出现 `Session ID unknown`	握手和升级落入不同 Worker，或旧 `sid` 在 Worker 重启后失效	对照 `engine.connection`、`engine.request`、`engine.upgrade.request` 的 `sid` 与 PID；并发发生时修正粘性，重启后则等待客户端重新连接。
WebSocket 升级失败或频繁回退 polling	客户端传输配置、代理路径、Upgrade 头或超时	先直连应用验证，再检查 `/socket.io/`、`proxy_http_version 1.1`、Connection 映射和代理超时
发送者收到回声，但其他客户端收不到广播	实际使用了 `socket.emit()`，或 Redis 适配器配置不一致	全局广播改用 `io.emit()`；检查所有 Worker 的 Redis 地址、认证、`REDIS_KEY` 和 `redis.adapter.ready` 日志
某些客户端正常，另一些完全收不到	客户端没有加入目标房间，或重连后未恢复成员关系	检查 `room:join` 回执、重连逻辑和房间消息测试
Worker 重启后客户端立即断开	长连接生命周期结果	区分正常发布、崩溃和内存重启；完善优雅退出、重连、重新鉴权和重新入房
重连成功，但房间和订阅消失	服务端状态绑定在旧 Socket 对象上	在 `connect` 后重新鉴权、恢复订阅并加入房间
Nginx 返回 502 或 504	PM2 上游不可用、监听地址错误、端口未发布或进程正在重启	检查 `0.0.0.0`、端口映射、`pm2 list`、容器健康状态和 upstream 配置
Worker 反复退出，内存持续增长	内存泄漏或连接对象未释放	结合 PM2 日志、内存排序和 `pm2 monit` 定位；`max_memory_restart` 只负责止损。