RRMQ

Redis Reliable Message Queue

Install

This library currently requires node@>4.0.0 and npm@>3.0.0. If you're using node@4.*, you probably have npm@2.* installed, then you need to run the following command first.

npm install -g npm@3

If you've setup you environment properly, run

npm install rrmq bluebird tuid ioredis

while bluebird, tuid and ioredis are peer dependencies of this library

Introduction

This library implements a message queue over redis based on feature of BRPOPLPUSH and PUBLISH/SUBSCRIBE, and is written in Typescript.

There are three role in this system:

• Producer

  An instance of producer may push message to queue

• Watchdog

  A watchdog subscribe a specified channel to watch the status of consumers. If one consumer is down, the message was being processed by that consumer will be recovered.

  There is a separated project rrmq-watchdog that provides a standalone daemon that you can just run it for your convenience.

• Consumer

  An instance of consumer await and pop message from the queue and meanwhile keeps heartbeat via publish message to a specified channel with a per-instance unique identifier to inform the watchdog that this instance is alive.

  If an instance died while processing a message. Watchdogs will notice that an instance has timed out and the message will be push back to the queue.

Example

• watchdog.js

const { RedisQueueWatchdog } = require('rrmq');
new RedisQueueWatchdog({
  watchdogRedisHost: 'localhost', watchdogRedisPort: 6379,
  redisHost: 'localhost', redisPort:6379,
  watchdogTopic: 'test-watchdog'
})
.on('error', console.error)
.start();

• consumer.js

const {RedisQueueConsumer} = require('rrmq');
new RedisQueueConsumer({
  watchdogRedisHost: 'localhost', watchdogRedisPort: 6379,
  redisHost: 'localhost', redisPort:6379,
  watchdogTopic: 'test-watchdog', queue: 'test-queue'
}).on('error', console.error)
.start(function(message) {
  return new Promise((done, fail)=> {
    // ...
  });
});

• producer.js

const {RedisQueueProducer} = require('rrmq');
new RedisQueueConsumer({
  redisHost: 'localhost', redisPort:6379,
  queue: 'test-queue'
}).on('error', console.error)
.send('hello world message');