Node.js多Worker节点下DATA对象一致性与请求串行化问询

一、解决数据一致性与请求顺序问题 #

你的核心问题在于两点：一是多Worker各自维护独立的DATA副本，导致修改不一致；二是RR调度下请求被随机分配，无法保证执行顺序。以下是两种可行方案：

方案1：单Worker接管所有数据变更请求 #

将PUT/DELETE这类写操作统一路由到同一个Worker处理，读请求可继续由多Worker分担。这样所有写操作串行执行，DATA仅存在一份副本，从根源避免一致性问题。

代码修改示例：

const cluster = require('cluster');
const http = require('http');
const numCPUs = require('os').cpus().length;
const { createProxyServer } = require('http-proxy'); // 需安装：npm i http-proxy

const proxy = createProxyServer({});
const MAIN_WORKER_PORT = 4000;
const WORKER_PORTS = [];

if (cluster.isMaster) {
  // 启动主Worker（处理写请求）
  const mainWorker = cluster.fork({ PORT: MAIN_WORKER_PORT });
  WORKER_PORTS.push(MAIN_WORKER_PORT);

  // 启动其他Worker（处理读请求）
  for (let i = 1; i < numCPUs; i++) {
    const port = MAIN_WORKER_PORT + i;
    WORKER_PORTS.push(port);
    cluster.fork({ PORT: port });
  }

  // 主进程作为统一入口，转发请求
  http.createServer((req, res) => {
    const isWriteRequest = (req.url === '/users' && (req.method === 'PUT' || req.method === 'DELETE'));
    if (isWriteRequest) {
      // 写请求固定转发到主Worker
      proxy.web(req, res, { target: `http://localhost:${MAIN_WORKER_PORT}` });
    } else {
      // 读请求轮询分配到其他Worker
      const targetPort = WORKER_PORTS[Math.floor(Math.random() * WORKER_PORTS.length)];
      proxy.web(req, res, { target: `http://localhost:${targetPort}` });
    }
  }).listen(3000); // 对外暴露的统一端口
} else {
  // Worker进程启动服务器
  const PORT = process.env.PORT;
  let DATA = require('./data.js'); // 加载本地DATA对象

  http.createServer((req, res) => {
    if (req.url === '/users' && req.method === 'PUT') {
      // 模拟耗时更新操作
      setTimeout(() => {
        // 假设从请求体解析更新内容
        const updateData = JSON.parse(req.body);
        DATA.users = { ...DATA.users, ...updateData };
        res.writeHead(200, { 'Content-Type': 'application/json' });
        res.end(JSON.stringify({ status: 'updated' }));
      }, 2000);
    } else if (req.url === '/users' && req.method === 'DELETE') {
      delete DATA.users;
      res.writeHead(200, { 'Content-Type': 'application/json' });
      res.end(JSON.stringify({ status: 'deleted' }));
    } else {
      // 处理读请求
      res.writeHead(200, { 'Content-Type': 'application/json' });
      res.end(JSON.stringify(DATA));
    }
  }).listen(PORT);
  console.log(`Worker running on port ${PORT}`);
}

方案2：队列化写请求+共享存储 #

如果必须用多Worker处理写操作，需将DATA迁移到共享存储（如Redis），同时用队列保证请求顺序。所有写请求先进入队列，Worker按顺序取出执行，Redis单线程特性天然保证操作顺序。

核心逻辑：

• 主进程将PUT/DELETE请求参数存入Redis列表队列
• 每个Worker监听队列，取出任务执行更新操作
• 读请求直接从Redis读取，无需排队

二、配置多端口Worker #

你当前代码中PORT++无法生效，因为每个Worker是独立进程，PORT变量在每个进程中都是初始值4000。正确做法是主进程fork时通过环境变量传递专属端口：

代码修改示例： #

const cluster = require('cluster');
const http = require('http');
const numCPUs = require('os').cpus().length;

if (cluster.isMaster) {
  // 为每个Worker分配不同端口
  for (let i = 0; i < numCPUs; i++) {
    const workerPort = 4000 + i;
    cluster.fork({ WORKER_PORT: workerPort });
  }
} else {
  // Worker从环境变量获取专属端口
  const PORT = process.env.WORKER_PORT;
  http.createServer((req, res) => {
    if(req.url === '/users' && req.method === "PUT") {
       updateUser(req, res); // 需处理响应，避免请求挂起
    } else if(req.url === '/users' && req.method === "DELETE") {
       deleteUser(req, res);
    } else {
      res.writeHead(404);
      res.end();
    }
  }).listen(PORT, () => {
    console.log(`Worker ${cluster.worker.id} listening on port ${PORT}`);
  });
}

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