将大数组传递给节点subprocess
我想要在大型arrays上执行复杂的CPU密集型工作。 理想情况下，我想把这个传递给subprocess。
var spawn = require('child_process').spawn; // dataAsNumbers is a large 2D array var child = spawn(process.execPath, ['/child_process_scripts/getStatistics', dataAsNumbers]); child.stdout.on('data', function(data){ console.log('from child: ', data.toString()); }); 但是当我这样做，节点给出的错误：
产卵E2BIG
我遇到了这篇文章
因此，将数据传输给subprocess似乎是要走的路。 我的代码现在是：
var spawn = require('child_process').spawn; console.log('creating child........................'); var options = { stdio: [null, null, null, 'pipe'] }; var args = [ '/getStatistics' ]; var child = spawn(process.execPath, args, options); var pipe = child.stdio[3]; pipe.write(Buffer('awesome')); child.stdout.on('data', function(data){ console.log('from child: ', data.toString()); });
然后在getStatistics.js中：
console.log('im inside child'); process.stdin.on('data', function(data) { console.log('data is ', data); process.exit(0); });
但process.stdin.on的callback没有达到。 如何在我的子脚本中接收stream？
编辑
我不得不放弃缓冲方法。 现在我发送数组作为消息：
var cp = require('child_process'); var child = cp.fork('/getStatistics.js'); child.send({ dataAsNumbers: dataAsNumbers });
但这只有在dataAsNumbers的长度低于20,000时才有效，否则超时。
有了这么大量的数据，我会考虑使用共享内存，而不是将数据复制到subprocess中（这是使用pipe道或传递消息时发生的情况）。 这样可以节省内存，缩短父进程的CPU时间，并且不太可能遇到一些限制。
shm-typed-array是一个非常简单的模块，看起来适合你的应用程序。 例：
parent.js
"use strict"; const shm = require('shm-typed-array'); const fork = require('child_process').fork; // Create shared memory const SIZE = 20000000; const data = shm.create(SIZE, 'Float64Array'); // Fill with dummy data Array.prototype.fill.call(data, 1); // Spawn child, set up communication, and give shared memory const child = fork("child.js"); child.on('message', sum => { console.log(`Got answer: ${sum}`); // Demo only; ideally you'd re-use the same child child.kill(); }); child.send(data.key);
child.js
"use strict"; const shm = require('shm-typed-array'); process.on('message', key => { // Get access to shared memory const data = shm.get(key, 'Float64Array'); // Perform processing const sum = Array.prototype.reduce.call(data, (a, b) => a + b, 0); // Return processed data process.send(sum); });
请注意，我们只是通过IPC向父进程发送一个小“钥匙”，而不是整个数据。 因此，我们节省了大量的记忆和时间。
当然，您可以将'Float64Array' （例如double ）更改为应用程序需要的任何types的数组 。 请注意，这个库特别只处理一维types的数组; 但这只是一个小小的障碍。
我也能够重现你所经历的延迟，但也许并不像你那么糟糕。 我用了以下
// main.js const fork = require('child_process').fork const child = fork('./getStats.js') const dataAsNumbers = Array(100000).fill(0).map(() => Array(100).fill(0).map(() => Math.round(Math.random() * 100))) child.send({ dataAsNumbers: dataAsNumbers, })
和
// getStats.js process.on('message', function (data) { console.log('data is ', data) process.exit(0) })
节点main.js 2.72s用户0.45s系统103％cpu 3.045总数
我生成100个由100个数字组成的元素来模拟你的数据，确保你正在使用message事件process 。 但也许你的孩子更复杂，可能是失败的原因，也取决于你设置的查询超时。
如果你想得到更好的结果，你可以做的是将你的数据分块成多个部分，将被发送到subprocess，并重build形成初始数组。
也有一种可能性是使用第三方库或协议，即使这样做有点多。 你可以看一下messenger.js ，甚至像AMQP队列，这样可以让你在两个进程之间进行通信，并保证subprocess确认消息。 它有几个节点的实现，像amqp.node ，但它仍然需要一些设置和configuration工作。
使用内存caching（如https://github.com/ptarjan/node-cache） ，并让父进程使用某个键存储数组内容，subprocess将通过该键检索内容。
您可以考虑使用操作系统pipe道， 您将在这里find要点作为节点子应用程序的input。
我知道这不是你所要求的，但是你可以使用集群模块（包含在节点中）。 通过这种方式，您可以获得与机器核心一样多的实例，以加快处理速度。 此外，如果在开始处理之前不需要获取所有可用数据，请考虑使用stream。 如果要处理的数据太大，我会将其存储在一个文件中，以便在过程中发生错误时重新进行处理。 这是一个聚类的例子。
var cluster = require('cluster'); var numCPUs = 4; if (cluster.isMaster) { for (var i = 0; i < numCPUs; i++) { var worker = cluster.fork(); console.log('id', worker.id) } } else { doSomeWork() } function doSomeWork(){ for (var i=1; i<10; i++){ console.log(i) } }
更多信息在工人问题8534462发送消息。
你为什么要做一个子过程？ 通过subprocess发送数据可能花费更多的CPU和实时时间，而不是在同一进程内进行处理。
相反，我build议对于超高效的编码，你可以考虑在工作线程中进行统计计算，这个工作线程和nodejs主进程在同一个内存中运行。
您可以使用NAN编写可以发布到工作线程的C ++代码，然后让该工作线程完成后将结果和事件发回到您的nodejs事件循环。
这样做的好处是，您不需要额外的时间将数据发送到不同的进程，但缺点是您将为线程操作编写一些C ++代码，但NAN扩展应该处理大多数对你来说是艰巨的任务。
对于长时间的工作任务，你可以使用像齿轮人员这样的工作，你可以在工作人员上做繁重的工作，这样就可以设置你需要的工作人员数量，例如我用这种方式进行文件处理，如果我需要规模，工人实例，我也有不同的工作人员为不同的任务，处理zip文件，生成缩略图等，这是好的工作人员可以写在任何语言node.js，Java，Python，并可以轻松地集成在您的项目
// worker-unzip.js const debug = require('debug')('worker:unzip'); const {series, apply} = require('async'); const gearman = require('gearmanode'); const {mkdirpSync} = require('fs-extra'); const extract = require('extract-zip'); module.exports.unzip = unzip; module.exports.worker = worker; function unzip(inputPath, outputDirPath, done) { debug('unzipping', inputPath, 'to', outputDirPath); mkdirpSync(outputDirPath); extract(inputPath, {dir: outputDirPath}, done); } /** * * @param {Job} job */ function workerUnzip(job) { const {inputPath, outputDirPath} = JSON.parse(job.payload); series([ apply(unzip, inputPath, outputDirPath), (done) => job.workComplete(outputDirPath) ], (err) => { if (err) { console.error(err); job.reportError(); } }); } function worker(config) { const worker = gearman.worker(config); if (config.id) { worker.setWorkerId(config.id); } worker.addFunction('unzip', workerUnzip, {timeout: 10, toStringEncoding: 'ascii'}); worker.on('error', (err) => console.error(err)); return worker; }
一个简单的index.js
const unzip = require('./worker-unzip').worker; unzip(config); // pass host and port of the Gearman server
我通常运行PM2的工人
与代码的集成非常简单。 就像是
//initialize const gearman = require('gearmanode'); gearman.Client.logger.transports.console.level = 'error'; const client = gearman.client(configGearman); // same host and port
只是将工作添加到传递函数名称的队列中
const taskpayload = {inputPath: '/tmp/sample-file.zip', outputDirPath: '/tmp/unzip/sample-file/'} const job client.submitJob('unzip', JSON.stringify(taskpayload)); job.on('complete', jobCompleteCallback); job.on('error', jobErrorCallback);