如何在Node.js的Worker Threads/Clusters间实现数据同步？

当Cluster的Worker需要主动修改共享数据并保持同步时，主线程单向分发的方式确实不适用，下面是几种可行的实现方案：

1. IPC 事件广播（轻量小数据首选） #

利用Cluster内置的IPC通信机制，让修改数据的Worker主动通知主进程，再由主进程转发给所有其他Worker，每个Worker维护一份数据副本。

主进程代码 #

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

if (cluster.isPrimary) {
  // 启动Worker进程
  for (let i = 0; i < numCPUs; i++) {
    cluster.fork();
  }

  // 监听Worker的广播请求，转发给其他Worker
  cluster.on('message', (senderWorker, msg) => {
    if (msg.type === 'SYNC_USER_LIST') {
      // 遍历所有Worker，跳过发送者
      Object.values(cluster.workers).forEach(worker => {
        if (worker.id !== senderWorker.id) {
          worker.send({
            type: 'UPDATE_USER_LIST',
            data: msg.data
          });
        }
      });
    }
  });
} else {
  // Worker进程逻辑见下方
}

Worker进程代码 #

const cluster = require('cluster');

// 每个Worker维护本地数据副本
let userList = [];

// 监听主进程转发的更新通知
process.on('message', (msg) => {
  if (msg.type === 'UPDATE_USER_LIST') {
    userList = msg.data;
    console.log(`Worker ${cluster.worker.id} 同步用户列表:`, userList);
  }
});

// 修改数据并触发广播的封装函数
function modifyUserList(newData) {
  userList = newData;
  // 通知主进程广播给其他Worker
  process.send({
    type: 'SYNC_USER_LIST',
    data: newData
  });
}

// 示例：Worker主动修改数据
modifyUserList(['Alice', 'Bob', 'Charlie']);

2. 外部共享存储（大数据/持久化场景首选） #

用第三方存储（如Redis、MongoDB）作为共享数据的唯一数据源，所有Worker直接读写这个外部存储，天然保证数据一致性。

Redis 示例代码 #

const redis = require('redis');
const cluster = require('cluster');

// 每个Worker连接同一个Redis实例
const redisClient = redis.createClient({
  host: 'localhost', // 根据实际配置修改
  port: 6379
});

// 获取最新用户列表
async function getUserList() {
  const rawData = await redisClient.get('shared_user_list');
  return rawData ? JSON.parse(rawData) : [];
}

// 修改用户列表并同步到Redis
async function updateUserList(newUser) {
  const currentList = await getUserList();
  currentList.push(newUser);
  await redisClient.set('shared_user_list', JSON.stringify(currentList));
  return currentList;
}

// Worker使用示例
updateUserList('David').then(updatedList => {
  console.log(`Worker ${cluster.worker.id} 更新后用户列表:`, updatedList);
});

// 其他Worker获取时直接调用getUserList即可拿到最新数据

注意区分 Cluster 和 Worker Threads #

Node.js的Cluster模块创建的是独立进程，无法直接共享内存；如果你的场景可以切换到Worker Threads（真正的线程），可以用SharedArrayBuffer配合Atomics实现内存级别的数据共享，但Cluster进程间不支持这种方式。

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