用于文件 I/O 操作的多核机器上的 Node.js

Question

我有点困惑，因为我读到的关于 Node cluster 模块的所有示例似乎只适用于网络服务器和并发请求。否则，对于 CPU 密集型应用程序，建议使用 worker_threads 模块。

那么 I/O 文件操作呢？想象一下，我有一个包含 100 万个文件名的数组：['1.txt', '2.txt', etc., ..., '1000000.txt']，我需要进行繁重的处理，然后写入结果文件内容？

有效使用 CPU 的所有内核将处理分散到不同文件名中的不同内核的方法是什么？

通常我会使用这个：

const fs = require('fs')
const fs = require('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？我应该如何使用它？

这行得通吗？

const fs = require('fs')
const fs = require('async')
const heavyProcessing = require('./heavyProcessing.js')

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

if (cluster.isPrimary) {
  console.log(`Primary ${process.pid} is running`);

  // Fork workers.
  for (let i = 0; i < numCPUs; i++) {
    cluster.fork();
  }

  cluster.on('exit', (worker, code, signal) => {
    console.log(`worker ${worker.process.pid} died`);
  });
} 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()
    })
  }
}

Answer 1

仅供大家了解，如果有兴趣，需要使用npm模块piscina。

在这个gist 我解释了一切。 NodeJS 是后端开发人员的强大工具，但您必须了解多核处理才能最大限度地发挥 CPU 的潜力。 这个 NodeJS 多核特性主要用于网络服务器，NodeJS 已经开箱即用了cluster 模块。 虽然 NodeJS 也有开箱即用的模块 threads，但处理起来并不容易。

让我们创建一个项目来测试单线程和多线程 CPU 密集型数据并将一些随机数据写入文件。

创建项目：

mkdir test-threads && cd test-threads
npm init -y

安装依赖并创建dist/目录

npm install async progress piscina command-line-args
mkdir dist

在项目目录根目录下创建文件index.js

const path = require('path')
const async = require('async')
const ProgressBar = require('progress')
const Piscina = require('piscina')
const commandLineArgs = require('command-line-args')

console.time('main')

const worker = require(path.resolve(__dirname, 'worker.js'))
const piscina = new Piscina({
  filename: path.resolve(__dirname, 'worker.js')
})

const argvOptions = commandLineArgs([
  { name: 'multi-thread', type: Boolean },
  { name: 'iterations', alias: 'i', type: Number }
])

const files = []
for (let i=0; i < (argvOptions.iterations || 1000); i++) {
  files.push(path.join(__dirname, 'dist', i + '.txt'))
}

var bar = new ProgressBar(':bar', { total: files.length, width: 80 });

async.each(files, function (file, cb) {
  (async function() {
    try {
      const err = argvOptions['multi-thread'] ? (await piscina.run(file)) : worker(file)
      bar.tick()
      if (err) cb(Error(err)); else cb()
    } catch(err) {
      cb(Error(err))
    }
  })();
}, (err) => {
  if (err) {
    console.error('There was an error: ', err)
    process.exitCode = 1
  } else {
    bar.terminate()
    console.log('Success')
    console.timeEnd('main')
    process.exitCode = 0
  }
})

现在在项目目录的根目录下创建worker.js

const fs = require('fs')

// some CPU intensive function; the higher is baseNumber, the higher is the time elapsed
function mySlowFunction(baseNumber) {
  let result = 0
  for (var i = Math.pow(baseNumber, 7); i >= 0; i--) {      
    result += Math.atan(i) * Math.tan(i)
  }
}

module.exports = (file) => {
  try {
    mySlowFunction(parseInt(Math.random() * 10 + 1))
    fs.writeFileSync(file, Math.random().toString())
    return null
  } catch (e) {
    return Error(e)
  }
}

现在只需在单线程上运行并检查经过的时间，进行 1000 次和 10000 次迭代（一次迭代等于数据处理和文件创建）

node index.js -i 1000
node index.js -i 10000

现在对比一下多线程的巨大优势

node index.js --multi-thread -i 1000
node index.js --multi-thread -i 10000

通过我所做的测试（16 核 CPU），差异是巨大的，它进行了 1000 次迭代，从单线程的 1:27.061 (m:ss.mmm) 到多线程的 8.884s。还要检查dist/ 中的文件以确保它们被正确创建。