Socket.io 扩展问题

Question

我有 3 个 Nodejs 应用程序在带有 Nginx 反向代理的 GCP 计算引擎实例（2cpu、2GB 内存、ubuntu 20.04）上运行。其中之一是 socket.io 聊天服务器。socket.io 应用程序@socket.io/cluster-adapter使用所有可用的 CPU 核心。\n我按照本教程更新了 Linux 设置以获得最大连接数。这是命令的输出ulimit，

\n

core file size          (blocks, -c) 0\ndata seg size           (kbytes, -d) unlimited\nscheduling priority             (-e) 0\nfile size               (blocks, -f) unlimited\npending signals                 (-i) 7856\nmax locked memory       (kbytes, -l) 65536\nmax memory size         (kbytes, -m) unlimited\nopen files                      (-n) 500000\npipe size            (512 bytes, -p) 8\nPOSIX message queues     (bytes, -q) 819200\nreal-time priority              (-r) 0\nstack size              (kbytes, -s) 8192\ncpu time               (seconds, -t) unlimited\nmax user processes              (-u) 7856\nvirtual memory          (kbytes, -v) unlimited\n

\n

cat /proc/sys/fs/file-max\n2097152\n

\n

/etc/nginx/nginx.conf

\n

user www-data;\nworker_processes auto;\nworker_rlimit_nofile 65535;\npid /run/nginx.pid;\ninclude /etc/nginx/modules-enabled/*.conf;\n\nevents {\n        worker_connections 30000;\n        # multi_accept on;\n}\n...\n

\n

/etc/nginx/sites-available/default

\n

...\n//socket.io part\nlocation /socket.io/ {\n         proxy_set_header X-Real-IP $remote_addr;\n         proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;\n         proxy_set_header Host $http_host;\n         proxy_set_header X-NginX-Proxy false;\n         proxy_pass http://localhost:3001/socket.io/;\n         proxy_redirect off;\n         proxy_http_version 1.1;\n         proxy_set_header Upgrade $http_upgrade;\n         proxy_set_header Connection "upgrade";\n        }\n...\n\n

\n

我的聊天服务器代码，

\n

const os = require("os");\nconst cluster = require("cluster");\nconst http = require("http");\nconst { Server } = require("socket.io");\nconst { setupMaster, setupWorker } = require("@socket.io/sticky");\nconst { createAdapter, setupPrimary } = require("@socket.io/cluster-adapter");\nconst { response } = require("express");\nconst PORT = process.env.PORT || 3001;\nconst numberOfCPUs = os.cpus().length || 2;\n\nif (cluster.isPrimary) {\n  const httpServer = http.createServer();\n  // setup sticky sessions\n  setupMaster(httpServer, {\n    loadBalancingMethod: "least-connection", // either "random", "round-robin" or "least-connection"\n  });\n\n  // setup connections between the workers\n  setupPrimary();\n  cluster.setupPrimary({\n    serialization: "advanced",\n  });\n\n  httpServer.listen(PORT);\n\n  for (let i = 0; i < numberOfCPUs; i++) {\n    cluster.fork();\n  }\n\n  cluster.on("exit", (worker) => {\n    console.log(`Worker ${worker.process.pid} died`);\n    cluster.fork();\n  });\n} \n//worker process\nelse {\n  const express = require("express");\n  const app = express();\n  const Chat = require("./models/chat");\n  const mongoose = require("mongoose");\n  const request = require("request"); //todo remove\n  var admin = require("firebase-admin");\n  var serviceAccount = require("./serviceAccountKey.json");\n  const httpServer = http.createServer(app);\n  const io = require("socket.io")(httpServer, {\n    cors: {\n      origin: "*",\n      methods: ["GET", "POST"],\n    },\n    transports: "websocket",\n  });\n\n  mongoose.connect(process.env.DB_URL, {\n    authSource: "admin",\n    user: process.env.DB_USERNAME,\n    pass: process.env.DB_PASSWORD,\n  });\n\n  app.use(express.json());\n\n  app.get("/", (req, res) => {\n    res\n      .status(200)\n      .json({ status: "success", message: "Hello, I\'m your chat server.." });\n  });\n\n  // use the cluster adapter\n  io.adapter(createAdapter());\n  // setup connection with the primary process\n  setupWorker(io);\n\n  io.on("connection", (socket) => {\n    activityLog(\n      "Num of connected users: " + io.engine.clientsCount + " (per CPU)"\n    );\n    ...\n    //chat implementations\n  });\n}\n

\n

负载测试客户端代码，

\n

const { io } = require("socket.io-client");\n\nconst URL = //"https://myserver.com/"; \nconst MAX_CLIENTS = 6000;\nconst CLIENT_CREATION_INTERVAL_IN_MS = 100;\nconst EMIT_INTERVAL_IN_MS = 300; //1000;\n\nlet clientCount = 0;\nlet lastReport = new Date().getTime();\nlet packetsSinceLastReport = 0;\n\nconst createClient = () => {\n  const transports = ["websocket"];\n\n  const socket = io(URL, {\n    transports,\n  });\n\n  setInterval(() => {\n    socket.emit("chat_event", {});\n  }, EMIT_INTERVAL_IN_MS);\n\n  socket.on("chat_event", (e) => {\n    packetsSinceLastReport++;\n  });\n\n  socket.on("disconnect", (reason) => {\n    console.log(`disconnect due to ${reason}`);\n  });\n\n  if (++clientCount < MAX_CLIENTS) {\n    setTimeout(createClient, CLIENT_CREATION_INTERVAL_IN_MS);\n  }\n};\n\ncreateClient();\n\nconst printReport = () => {\n  const now = new Date().getTime();\n  const durationSinceLastReport = (now - lastReport) / 1000;\n  const packetsPerSeconds = (\n    packetsSinceLastReport / durationSinceLastReport\n  ).toFixed(2);\n\n  console.log(\n    `client count: ${clientCount} ; average packets received per second: ${packetsPerSeconds}`\n  );\n\n  packetsSinceLastReport = 0;\n  lastReport = now;\n};\n\nsetInterval(printReport, 5000);\n\n

\n

正如您从代码中看到的，我仅用于websocket传输。因此，根据StackOverflow 的回答，它应该能够提供最多 8000 个连接。但是当我运行负载测试时，服务器在 1600 个连接后变得不稳定。CPU 使用率高达 90%，内存使用率高达 70%。我在 Nginx 错误日志中找不到\xe2\x80\x99 任何内容。如何将连接数增加到至少8000个？我应该升级实例或更改任何 Linux 设置吗？任何帮助，将不胜感激。

\n

更新\n我删除了与集群相关的所有内容，并将其作为常规单线程 Nodejs 应用程序再次运行。这次结果好一点，2800个稳定连接（CPU占用40%，内存占用50%）。请注意，我在测试期间没有执行任何磁盘 I/O。

\n

Answer 1

您正在使用集群适配器，该适配器不适合与粘性会话一起使用。您应该改用 Redis 适配器。每个工作人员都将连接到 Redis 并且能够相互通信。您还可以将 Redis 适配器与粘性会话一起使用，但您还需要在主进程上使用 Redis 适配器。

回答你的另一个问题：

“如果我删除粘性会话并仅使用 websocket，工作人员是否能够相互通信？”

是的，工人将能够相互沟通。我认为在聊天应用程序中使用粘性会话不是一个好主意。您应该使用 Redis 或 NATS 等发布/订阅系统在工作人员之间进行通信。例如，您可以使用Redis将消息发布到通道，其他worker将接收该消息并将其发送到客户端。

当您使用粘性会话时，每个工作人员将连接到一个客户端。因此，如果您有 4 名工作人员，您将能够同时为 4 名客户提供服务。如果您使用集群适配器，每个工作线程将连接到所有客户端。因此，如果您有 4 个工作人员，您将能够同时为 4 * 4 个客户提供服务。因此，您将能够使用集群适配器为更多客户端提供服务。

使用 Redis 适配器的示例：

const { createAdapter } = require("socket.io-redis");
const io = require("socket.io")(httpServer, {
  cors: {
    origin: "*",
    methods: ["GET", "POST"],
  },
  transports: "websocket",
});

io.adapter(createAdapter("redis://localhost:6379")); 

使用 NATS 适配器的示例：

const { createAdapter } = require("socket.io-nats");
const io = require("socket.io")(httpServer, {
  cors: {
    origin: "*",
    methods: ["GET", "POST"],
  },
  transports: "websocket",
});

io.adapter(createAdapter("nats://localhost:4222"));

尝试这两个选项，看看哪个最适合您。