Node.js多核心机器处理百万文件：cluster还是worker_threads？

场景与困惑 #

我查到的Node.js cluster模块示例几乎都是Web服务器和并发请求场景，而CPU密集型应用普遍推荐用worker_threads模块。那如果是文件IO+CPU密集的混合场景该怎么处理？

具体场景：我有一个包含100万个文件名的数组['1.txt', '2.txt', ..., '1000000.txt']，需要对每个文件名先执行CPU密集的heavyProcessing计算，再把结果写入对应文件。我想高效利用CPU所有核心，把不同文件的计算任务分配到不同核心上。

原始实现代码 #

const fs = require('fs')
const async = require('async') // 注：原代码重复require了fs，这里修正为async
const heavyProcessing = require('./heavyProcessing.js')

const files = ['1.txt', '2.txt', ..., '1000000.txt']

async.each(files, function (file, cb) {
  fs.writeFile(file, heavyProcessing(file), function (err) {
    if (!err) cb()
  })
})

疑问与尝试代码 #

我应该用cluster还是worker_threads？我写了下面的cluster代码，是否可行？

const fs = require('fs')
const async = require('async') // 注：原代码重复require了fs，这里修正为async
const heavyProcessing = require('./heavyProcessing.js')

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

if (cluster.isPrimary) {
  console.log(`主进程 ${process.pid} 正在运行`);

  // 启动对应CPU核心数的工作进程
  for (let i = 0; i < numCPUs; i++) {
    cluster.fork();
  }

  cluster.on('exit', (worker, code, signal) => {
    console.log(`工作进程 ${worker.process.pid} 已退出`);
  });
} else {
  const files = ['1.txt', '2.txt', ..., '1000000.txt']

  async.each(files, function (file, cb) {
    fs.writeFile(file, heavyProcessing(file), function (err) {
      if (!err) cb()
    })
  })
}

解答 #

先明确：你写的cluster代码不可行 #

你的核心问题是每个工作进程都在处理全部100万个文件，这会导致完全重复的计算，不仅没利用好多核，反而浪费系统资源。正确的逻辑应该是主进程负责拆分并分发任务，每个工作进程只处理分配给自己的那部分文件。

选型分析：cluster vs worker_threads #

两种模块都能满足你的需求，核心区别如下：

• cluster：基于多进程，每个worker是独立的Node.js进程，内存不共享，隔离性强，某个worker崩溃不会影响其他进程，适合纯CPU密集型任务，但进程间通信开销略大。
• worker_threads：基于多线程，共享主进程内存空间（有线程安全机制），线程间通信开销更小，适合需要共享部分数据的场景，同样能充分利用多核。

你的场景中两种方案都适用，下面分别给出正确实现：

方案一：使用cluster模块实现 #

核心逻辑：主进程拆分文件数组为多个分片，将每个分片分配给对应的worker进程，worker处理完自己的分片后通知主进程并退出。

const cluster = require('node:cluster');
const numCPUs = require('node:os').cpus().length;
const process = require('node:process');
const files = ['1.txt', '2.txt', ..., '1000000.txt'];

if (cluster.isPrimary) {
  console.log(`主进程 ${process.pid} 启动`);

  // 拆分文件数组为对应CPU核心数的分片
  const fileChunks = [];
  const chunkSize = Math.ceil(files.length / numCPUs);
  for (let i = 0; i < files.length; i += chunkSize) {
    fileChunks.push(files.slice(i, i + chunkSize));
  }

  // 启动worker并分配任务
  let chunkIndex = 0;
  for (let i = 0; i < numCPUs; i++) {
    const worker = cluster.fork();
    worker.send({ task: fileChunks[chunkIndex++] });

    // 监听worker完成信号
    worker.on('message', (msg) => {
      if (msg.type === 'done') {
        console.log(`工作进程 ${worker.process.pid} 已完成任务`);
        worker.kill();
      }
    });
  }

  cluster.on('exit', (worker) => {
    console.log(`工作进程 ${worker.process.pid} 已退出`);
  });
} else {
  // 工作进程逻辑
  const fs = require('fs');
  const async = require('async');
  const heavyProcessing = require('./heavyProcessing.js');

  process.on('message', (msg) => {
    const taskFiles = msg.task;
    // 控制文件IO并发数，避免系统过载
    async.eachLimit(taskFiles, 50, (file, cb) => {
      const result = heavyProcessing(file);
      fs.writeFile(file, result, (err) => {
        if (err) {
          console.error(`写入文件 ${file} 失败:`, err);
        }
        cb(err);
      });
    }, (err) => {
      if (err) {
        console.error(`工作进程 ${process.pid} 处理任务出错:`, err);
      }
      process.send({ type: 'done' });
    });
  });
}

方案二：使用worker_threads模块实现 #

核心逻辑：创建对应CPU核心数的线程，每个线程处理一个文件分片，线程通过workerData接收任务，完成后通知主线程。

const { Worker, isMainThread, parentPort, workerData } = require('node:worker_threads');
const numCPUs = require('node:os').cpus().length;
const files = ['1.txt', '2.txt', ..., '1000000.txt'];

if (isMainThread) {
  console.log(`主线程 ${process.pid} 启动`);

  // 拆分文件数组为对应CPU核心数的分片
  const fileChunks = [];
  const chunkSize = Math.ceil(files.length / numCPUs);
  for (let i = 0; i < files.length; i += chunkSize) {
    fileChunks.push(files.slice(i, i + chunkSize));
  }

  // 创建线程并分配任务
  fileChunks.forEach((chunk) => {
    const worker = new Worker(__filename, {
      workerData: { files: chunk }
    });

    worker.on('message', () => {
      console.log(`线程 ${worker.threadId} 已完成任务`);
    });

    worker.on('error', (err) => {
      console.error(`线程 ${worker.threadId} 出错:`, err);
    });

    worker.on('exit', (code) => {
      console.log(`线程 ${worker.threadId} 已退出，退出码 ${code}`);
    });
  });
} else {
  // 工作线程逻辑
  const fs = require('fs');
  const async = require('async');
  const heavyProcessing = require('./heavyProcessing.js');
  const { files } = workerData;

  async.eachLimit(files, 50, (file, cb) => {
    const result = heavyProcessing(file);
    fs.writeFile(file, result, (err) => {
      if (err) {
        console.error(`写入文件 ${file} 失败:`, err);
      }
      cb(err);
    });
  }, (err) => {
    if (err) {
      console.error(`线程 ${parentPort.threadId} 处理任务出错:`, err);
    }
    parentPort.postMessage({ type: 'done' });
  });
}

额外优化建议 #

1. 控制IO并发数：用async.eachLimit或类似工具限制同时写文件的数量（建议50-100），避免系统IO资源耗尽。
2. 错误重试机制：对写入失败的文件记录日志，后续可批量重试。
3. 内存监控：如果heavyProcessing生成的结果数据量较大，需监控进程/线程内存使用，避免内存溢出。

内容的提问来源于stack exchange，提问作者João Pimentel Ferreira