1

My main Node instance forks a worker process, which accepts messages over IPC (using the built-in Node process.send() and process.on('message'...) which are objects containing information about new jobs to add to Kue. It then processes those jobs.

My main Node instance calls something like this:

worker.send({jobType:'filesystem', operation: 'delete', path: fileDir});

and the worker instance does something like this:

jobs.create(message.jobType, message).save();

jobs.process('filesystem', function(job, done) {
    fs.delete(job.data.path, function(err) {
        done(err);
    });
});

and the job completes successfully.

How can I get callback-like functionality in my main Node instance when the job is completed? How can I return some results to the main Node instance from the worker instance?

4

1 回答 1

1

我相信我解决了这个问题，但如果有人可以改进我的解决方案或提供更好的解决方案，我会留下未解决的问题。

当您使用 Kue 在单独的进程中处理作业时，您不能简单地在作业完成时执行回调。这是两个进程之间通信的示例。我本来希望使用 Kue 自动提供每个作业的 id（我相信它与 Redis 中接收到的 id 相同）但是 app.js 需要在将作业的 id 发送给工作人员之前知道它收到消息时可以匹配id。

应用程序.js

var child = require('child_process');
var async = require('async');

var worker = child.fork("./worker.js");

//When a message is received, search activeJobs for it, call finished callback, and delete the job
worker.on('message', function(m) {
    for(var i = 0; i < activeJobs.length; i++) {
        if(m.jobId == activeJobs[i].jobId) {
            activeJobs[i].finished(m.err, m.results);
            activeJobs.splice(i,1);
            break;
        }
    }
});

// local job system
var newJobId = 0;
var activeJobs = [];

function Job(input, callback) {
    this.jobId = newJobId;
    input.jobId = newJobId;
    newJobId++;
    activeJobs.push(this);

    worker.send(input);

    this.finished = function(err, results) {
        callback(err, results);
    }
}


var deleteIt = function(req, res) {
    async.series([
        function(callback) {
            // An *EXAMPLE* asynchronous task that is passed off to the worker to be processed
            // and requires a callback (because of async.series)
            new Job({
                jobType:'filesystem',
                title:'delete project directory',
                operation: 'delete',
                path: '/deleteMe'
            }, function(err) {
                callback(err);
            });
        },
        //Delete it from the database
        function(callback) {
            someObject.remove(function(err) {
                callback(err);
            });
        },
    ],
    function(err) {
        if(err) console.log(err);
    });
};

worker.js

var kue = require('kue');
var fs = require('fs-extra');

var jobs = kue.createQueue();

//Jobs that are sent arrive here
process.on('message', function(message) {
    if(message.jobType) {
        var job = jobs.create(message.jobType, message).save();
    } else {
        console.error("Worker:".cyan + " [ERROR] No jobType specified, message ignored".red);
    }
});

jobs.process('filesystem', function(job, done) {

    if(job.data.operation == 'delete') {
        fs.delete(job.data.path, function(err) {
            notifyFinished(job.data.jobId, err);
            done(err);
        });
    }
});

function notifyFinished(id, error, results) {
    process.send({jobId: id, status: 'finished', error: error, results: results});
}

https://gist.github.com/winduptoy/4991718

于 2013-02-20T01:04:39.393 回答