Redis queue in Node.js cluster environment

Question

Using Node.js cluster support, I would like the master process to produce data items that are added to a Redis queue. I would then like to run multiple worker processes that read the Redis queue. Naturally, only one of the worker processes should consume the data item retrieved from the queue.

To get me started, can you suggest where to look in terms of Node package support or raw Redis commands to accomplish this? Let me emphasize the consumers are separate processes in a Node.js cluster environment, where we may adjust the number of worker processes competing to read from the single Redis queue to tune overall system performance.

Answer 1

It looks like kue is just what we need.

Minimal test program queuing tasks via Redis in a Node.js clustered environment:

// cluster-queue.js

var kue     = require('kue');
var cluster = require('cluster');

var numWorkers  = process.argv[2];
var numParallel = process.argv[3];
var jobDelay    = process.argv[4];
var numJobs     = process.argv[5];

if (process.argv.length !== 6) {
  console.log('Usage: node cluster-work-queue <numWorkers> <numParallel> <jobDelay> <numJobs>');
  process.exit(1);
}

var jobs = kue.createQueue();

var jobType = 'someTask';

if (cluster.isMaster) {
  for (var i = 0; i < numWorkers; i++) {
    cluster.fork();
  }
  for (var j = 0; j < numJobs; j++) {
    (function(count) {
      var data = { count: count };
      jobs.create(jobType, data).save(function(err) {
        if (err) {
          console.log('jobs.create', err);
        } else {
          console.log('master', 'jobs.create', count);
        }
      });
    })(j);
  }
} else {
  jobs.process(jobType, numParallel, function(job, done){
    console.log('worker', cluster.worker.id, 'jobs.process', job.data);
    setTimeout(function() { done(); }, jobDelay);
  });
}

Sample run (note command line options to adjust worker counts, etc.):

[jfathman@cloud node]$ node cluster-queue.js 2 1 20 10
master jobs.create 0
master jobs.create 1
master jobs.create 2
master jobs.create 3
master jobs.create 4
master jobs.create 5
master jobs.create 6
master jobs.create 7
master jobs.create 8
master jobs.create 9
worker 1 jobs.process { count: 0 }
worker 2 jobs.process { count: 1 }
worker 1 jobs.process { count: 2 }
worker 2 jobs.process { count: 3 }
worker 1 jobs.process { count: 4 }
worker 2 jobs.process { count: 5 }
worker 1 jobs.process { count: 6 }
worker 2 jobs.process { count: 7 }
worker 1 jobs.process { count: 8 }
worker 2 jobs.process { count: 9 }
^C