Hi, I was trying to produce multiple messages and encountered this warning

(node:1055180) MaxListenersExceededWarning: Possible EventEmitter memory leak detected. 11 up listeners added to [Producer]. Use emitter.setMaxListeners() to increase limit
    at _addListener (node:events:601:17)
    at Producer.addListener (node:events:619:10)
    at Producer.once (node:events:663:8)
    at Producer.produce (node_modules/redis-smq/dist/src/lib/producer/producer.js:94:26)

It's from the once(events.UP, proceed).

• 我使用的是Egg框架，delay.worker可以收到到其他线程发送的消息，你应该加个消息判断
• I am using the Egg framework, delay.worker can receive messages sent by other threads, you should add a message to judge

Hello

The latest version of Redis allows for Promises and is a much nicer way of working, at the moment we use Handi-Redis to bypass this as using Redis 4 conflicts with this package's Redis 3.*.

I will look forward to your reply

Best Regards Chris

This project is very interesting and straight forward to use.

One use-case I'd like is to throttle message consumption for the purpose of consuming 3rd party API calls. It would be good to have an configuration for the consumer which allows you to set the interval and quantity within that interval, much like this library: https://www.npmjs.com/package/p-throttle

Is this something you would be interested in adding?

I'm trying to figure out how to add delay between retry on error. Neither setScheduledPeriod or setScheduledDelay seems to do it. I want to retry a .job n number of times only on failure and there must be 10 minutes delay between retry. Is this possible to do?

Hi there My consumer may have many message handlers for different queues. By setting up new message handlers, there might be errors because the handlers already exist.

So I want to check if a message handler already exists before adding a new one. Is it possible to expose getMessageHandler or there is a better approach?

Plus, if I have multiple instances of the same app which use consumers, they actually won't know each other's messageHandler? In that case, is it possible to find out any consumer is consuming a particular queue (and ns)?

Hi there

const { Consumer, Message, Producer } = require('redis-smq');

const consumer = new Consumer();

const messageHandler = (msg, cb) => {
  const payload = msg.getBody();
  console.log('handler1: ', payload);
  cb(); // acknowledging the message
};

// the second parameter is for enabling priority queuing
consumer.consume('queue1', false, messageHandler, (err, isRunning) => {
  if (err) console.error(err);
  // the message handler will be started only if the consumer is running
  else {
    console.log(
      `handler1 has been registered. Running status: ${isRunning}`,
    );
  } // isRunning === false
});

const anotherMessageHandler = (msg, cb) => {
  const payload = msg.getBody();
  console.log('handler2: ', payload);
  // ...
  cb();
};

consumer.consume(
  'queue2',
  false,
  anotherMessageHandler,
  (err, isRunning) => {
    if (err) console.error(err);
    // the message handler will be started only if the consumer is running
    else {
      console.log(
        `handler2 has been registered. Running status: ${isRunning}`,
      );
    }
  },
);

consumer.run();

// -------------------------------------------------

const message1 = new Message();
message1
  .setBody({ hello: 'world1' })
  .setTTL(3600000) // in millis
  .setQueue('queue1');

const message2 = new Message();
message2
  .setBody({ hello: 'world2' })
  .setTTL(3600000) // in millis
  .setQueue('queue2');

const producer = new Producer();
producer.produce(message1, (err) => {
  if (err) console.log(err);
  else {
    const msgId = message1.getId(); // string
    console.log('Successfully produced. Message ID is ', msgId);
  }
});


producer.produce(message2, (err) => {
  if (err) console.log(err);
  else {
    const msgId = message2.getId(); // string
    console.log('Successfully produced. Message ID is ', msgId);
  }
});

For the code above, both queues are not priority queue. The output will be always:

handler1 has been registered. Running status: false
handler2 has been registered. Running status: false
Successfully produced. Message ID is  8ce2fca4-5e41-49cb-9de6-cd3eedcbaad0
Successfully produced. Message ID is  69b7f2e1-a532-4823-a4b4-88f1a42102e5
handler1:  { hello: 'world1' }
handler2:  { hello: 'world2' }

If I change the the second one to priority queue consumer.consume('queue2', true, anotherMessageHandler,..., I run it for a few times, I always get this result:

handler1 has been registered. Running status: false
handler2 has been registered. Running status: false
Successfully produced. Message ID is  736395df-bd62-4762-8176-5fb96d857cbc
Successfully produced. Message ID is  ce50472d-2926-4f25-9d7a-bd19ed6b5e52
handler1:  { hello: 'world1' }

If I set priority back to false for the second queue, I got this output:

handler1 has been registered. Running status: false
handler2 has been registered. Running status: false
Successfully produced. Message ID is  1f14a3f4-71ee-4eec-8d4b-938b0883c4a9
Successfully produced. Message ID is  c8f6e2c7-e497-4902-88d5-78f38c7d2ab3
handler1:  { hello: 'world1' }
handler2:  { hello: 'world2' }
handler2:  { hello: 'world2' }
handler2:  { hello: 'world2' }
handler2:  { hello: 'world2' }
handler2:  { hello: 'world2' }
handler2:  { hello: 'world2' }
handler2:  { hello: 'world2' }
handler2:  { hello: 'world2' }

My question is, when I use consumer.consume('queue2', true, anotherMessageHandler,..., why couldn't I consume the messages from 2nd queue?

Hi there.

While consumer is running, when I call consumer.shutdown() twice, the second call throws Can not switch state while another state transition is in progress. error.

Actual:

     consumer.shutdown();
     try {
        consumer.shutdown();
      } catch (err) {
       // error caught as "Can not switch state while another state transition is in progress."
        console.error(err.message);
      }

Actual:

    consumer.shutdown();
    consumer.shutdown((err)=>{
        // unable to catch "Can not switch state while another state transition is in progress." error
    });

IMO, any error should be thrown to the callback in shutdown(cb).

Expect:

    consumer.shutdown();
    consumer.shutdown((err)=>{
        // catch error here
    });

Hi there, with Producer, I'm able to produce/consume the message, but with MultiQueueProducer, it seems it doesn't work.

This works

const { Consumer, Producer, Message } = require('redis-smq');
const queue = 'S-12345'

const message = new Message();
message
    .setBody({ hello: 'world', ts: new Date() })
    .setTTL(3600000);

const producer = new Producer(queue);

setInterval(() => {
    producer.produce(message, (err) => {
        if (err) console.log(err);
        else console.log('Successfully produced')
    });
}, 1000);

class TestQueueConsumer extends Consumer {
    consume(message, cb) {
        console.log('Got a message to consume:', message.body);
        cb();
    }
}

const consumer = new TestQueueConsumer(queue);
consumer.run();
console.log('consumer running')

This doesn't work

const { Consumer, MultiQueueProducer, Message } = require('redis-smq');

const queue = 'S-12345'

const message = new Message();
message
    .setBody({ hello: 'world', ts: new Date() })
    .setTTL(3600000);

const producer = new MultiQueueProducer();

setInterval(() => {
    producer.produce(queue, message, (err) => {
        if (err) console.log(err);
        else console.log('Successfully produced')
    });
}, 1000);

class TestQueueConsumer extends Consumer {
    consume(message, cb) {
        console.log('Got a message to consume:', message.body);
        cb();
    }
}

const consumer = new TestQueueConsumer(queue);
consumer.run();
console.log('consumer running')

Anything wrong? Thanks in advance.

Hi there, In my use case, I have to send the same message to a large number of queues. By using Producer class, I will have to create multiple instances of producers for multiple queues.

What's the more efficient way to do this? e.g. can we have something like producer.produce(queue, message, cb)?

Thanks.

If we are talking about high performance, then a large number of connections create locks at the redis level, and in our case, these are 15 connections per 1 Producer and 1 consumer: Consumer - 11 connections Producer - 4 connections

I researched the code and saw no reason to use that many connections. For a producer, 1 connection is enough. For the consumer, 3 connections are enough, one for the main process and two for the forks.

Actually all ioredis configuration parameters are supported.

But there is no way to provide and use a cluster configuration.

See https://github.com/luin/ioredis#cluster

hi there, Is it possible to limit the size of the queue. For example, I'm only interested in last 10 messages of a queue. The queue will hold 10 messages max and old messages will be dumped.

This is common for some queue systems or databases.

Not super important but nice to have.

@weyoss Thanks for nicely crafted library.

I have done some performance benchmark to compare redis-smq with bull on my Dell XPS. Machine 16GB RAM, Hexa Core

No doubt, redis-smq is clear winner because of it's simplicity and focused approach towards message queuing.

Following is the messages produced and consumed rate | Scenarios | redis-smq produce rate and consumed rate | bull produce rate and consumed rate | |--- |--- |--- | | 1 producer and 1 consumer | 16.5k/sec and 9k/sec | 9.5k/sec and 7k/sec | | 1 producer and 2 consumer | 15k/sec and 13.5k/sec | 5.0k/sec and 5.0k/sec | | 2 producer and 2 consumer | 23.3k/sec and 13.2k/sec | 13.3k/sec and 10.6k/sec | | 4 producer and 4 consumer | 30k/sec and 17.8k/sec | 13.4k/sec and 10.4k/sec | | 6 producer and 3 consumer | 40k/sec and 16.1k/sec | 17.5k/sec and 12.2k/sec |

It is clearly visible that bull is almost saturated in terms of producer and consumer rate. There can be multiple reasons of less performance in bull like promises or multiple checks & conditions to provide other features like rate limiting etc.

Though, I am not able to convince myself which library to use for my next project, because redis-smq simplicity and performance is far ahead of bull, which is the purpose of this lib, but bull has lot of features, which can be useful when things are in production.

I am thinking to move from RabbitMQ to Redis, because I don't need complex routing like RabbitMQ. I have also seen heavy usage for simple queuing task in RabbitMQ, where RabbitMQ was taking approx 2GB for holding 1 million messages and total messages size was around 100MB. So ideally in Redis, it would be 100MB only. I understand RabbitMQ do lot of things out of the box, but I think Redis can work very well in simple queuing cases.

Any suggestion would be appreciated by anyone to take this decision. I am also interested to listen more from actual user of this library who have used it in production with heavy load.

Thanks.