Having a Asynchronous Task Queue Using Redis and Kue.js

This a tutorial of how to set up a redis as a message queue broker using Docker and having a task queue using Kue.js on Node.js environment.

Photo by Melanie Pongratz on Unsplash

Here’s a traditional Chinese version of this post:

Having a Asynchronous Task Queue Using Redis and Kue.js

Outline

1. Deploy a redis database using Docker.
2. Introduction to Kue.js

Deploy Redis

Prepare a shell script with following docker commands. It’s more easy to use later if we put commands in a shell script.

#!/bin/shdocker run                        \
  -d                              \
  -p 6379:6379                    \
  -v $PWD/redis_data:/data        \
  --name my-redis                 \
  redis

Let me explain what this command actually do.

1. -d, running in the background
2. -p 6379:6379, mapping port:6379 from docker container to localhost port:6379
3. -v $PWD/redis_data:/data, container will use the /redis_data in current directory as his /data directory. If we remove this container or it becomes broken due to some reasons, the data are still there.
4. redis is the image name. we can add a tag like redis:5.0, redis:latest to assign what version you want to use and default is latest. Wanna know what version are available you can check this: https://hub.docker.com/_/redis/

After running the command above, you can run docker ps to check. If success, you will see following descriptions.

Introduction to Kue.js

Install

$ npm install kue

Connection

var kue = require('kue')
var queue = kue.createQueue();

It will connect to localhost:6379 by default. If you want to change:

var queue = kue.createQueue({
  redis: {
    port: 1234,
    host: '10.0.50.20',
    auth: 'password',
    db: 3, // if provided select a non-default redis db
    options: {
      // see https://github.com/mranney/node_redis#rediscreateclient
    }
  }
});

Produce message

After connection, we can produce messages~

queue.create('email', {
    title: 'welcome email for tj',
    to: 'tj@learnboost.com',
    template: 'welcome-email',
}).save( function(err){
   if( !err ) console.log( job.id );
});

Here, we produce a message to a queue named email with a JOSN object payload.

Consume message

Now, we can consume the messages we produced in other places by using queue.process function.

queue.process('email', function(job, done){
  doSomethingWithData(job.data)
  
  done();//acknowledge this message
});

queue.process takes two parameters. The first one is the name of queue which is email here for example. The second one is a callback function taking two parameters too. The first is job and job.data is exactly the message we sent above, and the second one is a callback function done. When we finish processing with that message we can call done() to acknowledge.

Concurrency controll

In the above example, we can only consume one message at a time. If your job is IO intensive, you may want to process more message concurrently.Now you can call process with one more parameters, and the second one is concurrency.

queue.process('email', 10, function(job, done){
  doSomethingWithData(job.data)
  
  done();//acknowledge this message
});

For now, we have deployed a redis database server and use Kue.js to produce and consume messages. To learn more about Kue.js, check it’s github:

Automattic/kue