weyoss / redis-smq Goto Github PK
View Code? Open in Web Editor NEWA simple high-performance Redis message queue for Node.js.
License: MIT License
A simple high-performance Redis message queue for Node.js.
License: MIT License
Is it possible?
thanks
We recently updated our redis-smq Node package from 1.0.22 to 1.1.1, and our consumers and producers started to fail because it would attempt to connect to 127.0.0.1 instead of the address specified in our config. Our config file looks like this:
module.exports = {
namespace: 'dpp-redis-server',
redis: {
host: 'our.host.here',
port: 6379
},
log: {
enabled: 1
},
monitor: {
enabled: true,
host: '127.0.0.1',
port: 3000,
},
};
We noticed that the example config file in the repo has a slightly different configuration than version 1.0.22. Once we updated the config file, then it works as expected.
However, the biggest issue is that npm updates from 1.0.22 to 1.1.1 without expecting breaking API changes.
Could you look into adding backwards compatibility between configs from version 1.0 into 1.1?
Need to stop the producer without execute "process.exit(0);"
Hello everybody,
i had a problem when i run monitor service.
PS D:\server\chat-server> node .\monitor-queue.js
Monitor server is running...
D:\server\chat-server\node_modules\redis-smq\src\stats-frontend.js:204
if (err) throw err;
^
ReplyError: ERR wrong number of arguments for 'set' command
at parseError (D:\server\chat-server\node_modules\redis-parser\lib\parser.js:193:12)
at parseType (D:\server\chat-server\node_modules\redis-parser\lib\parser.js:303:14)
I am using node version v7.10.1.
Please help me!
Thank
First of all I must say I'm super happy with the performance of this Plugin. I would like to make an open question regarding redis-smq.
In the admin dashboard I'm working on, I'd like to have the hability of displaying the amount of running tasks, failed queues, etc.
I tried with a simple redis client, and trying to search on the following keys:
const keyTypes = {
KEY_TYPE_MESSAGE_QUEUE: '1.1',
KEY_TYPE_PROCESSING_QUEUE: '1.2',
KEY_TYPE_DEAD_LETTER_QUEUE: '1.3',
KEY_TYPE_MESSAGE_QUEUE_DELAYED: '1.4',
KEY_TYPE_HEARTBEAT: '2.1',
KEY_TYPE_GC_LOCK: '3.1',
KEY_TYPE_GC_LOCK_TMP: '3.2',
KEY_TYPE_RATE: '4',
KEY_TYPE_RATE_INPUT: '4.1',
KEY_TYPE_RATE_PROCESSING: '4.2',
KEY_TYPE_RATE_ACKNOWLEDGED: '4.3',
KEY_TYPE_RATE_UNACKNOWLEDGED: '4.4',
KEY_TYPE_STATS_FRONTEND_LOCK: '5.1',
KEY_TYPE_MESSAGE_QUEUES_INDEX: '6.1',
KEY_TYPE_PROCESSING_QUEUES_INDEX: '6.2',
KEY_TYPE_DEAD_LETTER_QUEUES_INDEX: '6.3',
KEY_TYPE_SCHEDULER_LOCK: '7.1',
KEY_TYPE_SCHEDULER_LOCK_TMP: '7.2',
};
Ideally I would like to understand how to search in this tree of keys, and avoid using a redis client. I could implement that part perfectly and would be happy to help but i'm lost on how finding the info.
Hi,
If a consumer throws a message before sending the acknowledgment then it is received after the consumer is restarted. But if the retry attempt is more than 3 then the message is deleted. What should be the value of setRetryThreshold
so that the message is not automatically deleted unless the consumer sends ACK? Have tried setting message.setRetryThreshold(-1)
but it doesn't work. Also, what is the default value of messageConsumeTimeout
?
//Producer.ts
'use strict';
const { Message, Producer } = require('redis-smq');
const message = new Message();
message
.setBody({hello: 'worldd!'})
.setTTL(10800000)
const config = {
namespace: 'test',
redis: {
driver: 'redis',
options: {
host: '127.0.0.1',
port: 6379,
connect_timeout: 3600000,
},
}
}
const producer = new Producer('test_queue',config); //test_queue
producer.produceMessage(message, (err) => {
if (err) console.log(err);
else console.log('Successfully produced')
});
//Consumer.ts
'use strict';
const redisSMQ = require('redis-smq');
const Consumer = redisSMQ.Consumer;
class TestQueueConsumer extends Consumer {
constructor() {
super(config)
}
consume(message, cb) {
try{
console.log('Got a message to consume:', message);
throw new Error("test")
//cb();
} catch(e) {
console.log("consumer error",e)
}
}
}
TestQueueConsumer.queueName = 'test_queue';
const consumer = new TestQueueConsumer();
consumer.run();
Hi,
After researching for all the available Redis queues, I found this to be the best for our use case :)
Since I haven't encountered any errors yet and it fulfills most of the use case.
I just wanted to clarify a few things, Let me know if this is not possible.
-I have multiple workers that process jobs, no 2 workers should process the same job.
-After a worker's job is complete the worker ACKs the job ( using cb() ). Until then the job is unavailable for other workers.
-If one of the apps terminates abruptly; e.g. via CTRL-C then the job should be sent back to the queue so that it can be picked by the next available worker. Which I think it does.
(correct me if it will not since that is the most important criteria)
However, I do not want to remove the message automatically from the queue. Messages should only be removed by the worker.
I tried doing this message.setBody(msg).setTTL(-1)
but it gives error.
Is there a way to do so. If not what is the maximum TTL that I can provide.
Hi,
We are running in a redundant redis setup with sentinels. When I kill the master redis(kill the VM), it takes a while for a redundant master to take over. During that time, I guess an error is sent by ioredis and, since that error is thrown by the redis-client, it never gets caught.
Is there a reason why there is no recovery after a transient redis error
this.client.once('error', (err: Error) => {
throw err;
});
Thanks
Instead of shutting down the producer if there is an elegant way to call a function and stop the published message that will be nice. This could be useful if the publish is triggered by the user action from the client side and can be useful stopping the previous publish and generate new one.
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?
It's not explicitly stated but the library appears to rely on Redis being configured for persistence. In other words: there is no way for us to provide our own persistence mechanism.
Our product is deployed to a large number of customers with Redis persistence disabled. I would LOVE to be able to just switch Redis persistence on, but updating deployments is... complicated and likely to break installations. I have reached out to other contributors and management for input on making this switch but, in the mean time, I'm looking at alternatives.
Would a PR that allows us to provide a "Persistence" class be accepted? The assumption is that the default class's functionality would always be a no-op and rely on Redis. In our case, we could provide an instance of this class through the configuration and the library would call the functions on that instance to persistence the data else-where.
I am trying to create alerts and/or display queue size in our internal monitoring/alerting/dashboard system. Is there any way to get the metrics from the queues other than through the monitoring web site?
Is there some support for recurring tasks, similar to Bull? I suspect I could use something like later, but it would be good to know...
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
});
Line 301 in 964f577
Should the above line call this[sProcessMessageFailure] instead of throwing error?
Situation: Our extended consumer class's consume method has a promise, so if any error occurs inside that promise, its being unhandled by the consume method (even though there's a catch) and onConsumed throws that error and halts the process.
Please guide.
Hi there, with Producer
, I'm able to produce/consume the message, but with MultiQueueProducer
, it seems it doesn't work.
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')
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.
The producer cannot check the existence of the message on restart and duplicates it.
There is no way to set a unique name for a cron task and update it instead of creating duplicates.
There is no way to delete a task that runs regularly.
Thus, the use of cron tasks loses all meaning, without being able to manage them and control their work.
Since upgrading to the latest version(2.0.11), I always get this error when I failover my redis server.
Error: Can not switch to the same current state.
at switchState (/opt/app-root/src/node_modules/redis-smq/src/power-state-manager.js:19:19)
at Object.goingUp (/opt/app-root/src/node_modules/redis-smq/src/power-state-manager.js:62:13)
at Object.start (/opt/app-root/src/node_modules/redis-smq/src/scheduler.js:219:31)
at Producer. (/opt/app-root/src/node_modules/redis-smq/src/instance.js:46:36)
at Producer.emit (events.js:314:20)
at Producer. (/opt/app-root/src/node_modules/redis-smq/src/instance.js:43:18)
at Producer.emit (events.js:314:20)
at Producer.completeBootstrap (/opt/app-root/src/node_modules/redis-smq/src/instance.js:264:14)
at Producer. (/opt/app-root/src/node_modules/redis-smq/src/instance.js:40:60)
at Producer.emit (events.js:314:20)
at /opt/app-root/src/node_modules/redis-smq/src/instance.js:195:23
at /opt/app-root/src/node_modules/neo-async/async.js:6883:13
at /opt/app-root/src/node_modules/redis-smq/src/instance.js:184:22
at tryCatcher (/opt/app-root/src/node_modules/standard-as-callback/built/utils.js:11:23)
at /opt/app-root/src/node_modules/standard-as-callback/built/index.js:19:49
at processTicksAndRejections (internal/process/task_queues.js:93:5)
Looks like the GOING_UP in the power-state-manager is emitted while it is already UP. I am using ioredis with sentinel support, the failover takes ~10sec.
I haven't seen this issue with version 2.0.0
Is there any plan to include a pause/ resume queue feature? May be on global level so no worker can consume from the queue until the queue is resumed.
Hi,
Is it possible to create new consumers on the fly during a running node process rather than have them run separately as a standalone CLI.
Excellent work on creating this SMQ.
Thanks,
Mahesh
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.
Currently, redis password is not used while creating redis connection. Need to support redis password and db.
Please suggest, should the getNewInstance (redis-client.js) be modified to include password and db or is there a better way to do it without touching this library files.
Thank you.
hi,
what is the best way to check if the que is not empty? should i create a socketio client like the one in monitor GUI?
In the consumer constructor, project namespace must be set (setNamespace()) before generating the keys (redisKeys.getKeys()).
Currently, while construing consumer this is causing the keys to be generated incorrectly with default namespace in it, like redis-smp-default. Rest of the times, getKeys generating correct keys.
Monitor needs namespace update in statsFrontend()
if (config.hasOwnProperty('namespace')) redisKeys.setNamespace(config.namespace);
In a lot of projects there are existing loggers. It'd be great to accept a bunyan compatible logger instance in the options and use that over making one internally. This will allow the host app to use their own levels and transports without duplicating those over to redis-mq.
if redis server doesn't start, run monitorServer.listen()
will get app crash
Line 93 in b35bef1
Was the ns intended to be redis-smp- instead of redis-smq ?
npm ERR! Windows_NT 10.0.15063
npm ERR! argv "C:\Program Files\nodejs\node.exe" "C:\Program Files\nodejs\node_modules\npm\bin\npm-cli.js" "install" "redis-smq" "--save"
npm ERR! node v7.0.0
npm ERR! npm v3.10.8
npm ERR! code ENOSELF
npm ERR! Refusing to install redis-smq as a dependency of itself
npm ERR!
npm ERR! If you need help, you may report this error at:
npm ERR! https://github.com/npm/npm/issues
npm ERR! Please include the following file with any support request:
npm ERR! C:\xxxxxxx\examples\nodejs\redis-smq\npm-debug.log
node -v = 7.0.0
Hi.
The monitoring not working in version 1.0.8, but his work in v1.0.7
TypeError: Cannot destructure property
queueName
of 'undefined' or 'null'.
at Object.getKeys (D:\nodeProjects\redissmqcollector\node_modules\redis-smq\src\redis-keys.js:14:29)
at statsFrontend (D:\nodeProjects\redissmqcollector\node_modules\redis-smq\src\stats-frontend.js:15:29)
at Server.server.listen (D:\nodeProjects\redissmqcollector\node_modules\redis-smq\src\monitor\index.js:52:31)
at Object.onceWrapper (events.js:313:30)
at emitNone (events.js:111:20)
at Server.emit (events.js:208:7)
at emitListeningNT (net.js:1378:10)
at _combinedTickCallback (internal/process/next_tick.js:135:11)
at process._tickCallback (internal/process/next_tick.js:180:9)
at Function.Module.runMain (module.js:678:11)
Thanks
Hi,
I have recently come across this module.
Does it allow setting message-id to a job like this https://github.com/bee-queue/bee-queue#jobsetidid
I tried bee-queue but it has some bugs which are not acceptable in our case.
Also, can I provide concurrency to consumers?
Are they available? I would like to try this package in my TS app.
Typescript can't compile because of an error.
In types/config.ts
interface ConfigInterface
extends Monitor.ConfigInterface
export interface ConfigInterface extends Monitor.ConfigInterface {
namespace?: string,
redis: {
driver: ConfigRedisDriver,
options: {
[key:string]: any
}
},
log?: {
enabled: boolean,
options: {
[key:string]: any
}
},
}
but redis
is not suitable for Monitor.ConfigInterface
interface, because of different types of driver
. They look equal, but that's different enums.
export enum ConfigRedisDriver {
REDIS = "redis",
IOREDIS = "ioredis"
}
and
export enum RedisDriver {
REDIS = 'redis',
IOREDIS = 'ioredis'
}
In version 2.0.0 there wasn't inheritance and there wasn't a typescript error. In 2.0.1 it was added.
events.js:183
throw er; // Unhandled 'error' event
^
ReplyError: ERR wrong number of arguments for 'set' command
at parseError (C:\Users\212702834\Documents\Userdata\Practice\redis-message-queue\node_modules\redis-parser\lib\parser.js:193:12)
at parseType (C:\Users\212702834\Documents\Userdata\Practice\redis-message-queue\node_modules\redis-parser\lib\parser.js:303:14)
Dependency Versions:
{
"redis": "2.7.1",
"redis-smq": "^1.0.20"
}
Running on Windows 10, x64
Redis server version 2.4.5
Steps to reproduce:
ps: redis is running with aof
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.
/home/aappana/nodejs/node_modules/redis-smq/src/monitor/index.js:33
app.use(async (ctx) => {
^
SyntaxError: Unexpected token (
at createScript (vm.js:56:10)
at Object.runInThisContext (vm.js:97:10)
at Module._compile (module.js:542:28)
at Object.Module._extensions..js (module.js:579:10)
at Module.load (module.js:487:32)
at tryModuleLoad (module.js:446:12)
at Function.Module._load (module.js:438:3)
at Module.require (module.js:497:17)
at require (internal/module.js:20:19)
at Object. (/home/aappana/nodejs/node_modules/redis-smq/index.js:5:17)
Ran this in linux - As this library doesnot support windows. your project architecture was good. but failed.
It would be helpful to have a configurable namespace instead of the default hard coded ns redis-smq, via the config file. Helpful in cases where redis-smq is a module of a larger redis based projects.
When a connection error occurs (Connection timetout), one of the events will end the main application and there is no way to prevent that.
There should be some handler that allowed whoever used the application to handle the failure and decide by himself if the application should be terminated.
File:
.on(events.ERROR, (err) => {
if (err.name !== 'AbortError' && state !== states.GOING_DOWN) {
loggerInstance.error(err);
process.exit(1);
}
});
How do I define the db function? Does this work?
const redisSMQ = require('redis-smq');
const ingestion = require('./ingestion');
const { Consumer } = redisSMQ;
class QueueConsumer extends Consumer {
// eslint-disable-next-line class-methods-use-this
consume(message, cb) {
ingestion.myfunction(message);
cb();
}
}
module.exports = { QueueConsumer };
And also how do we check if the queue is empty?
Hi!
My dead-letter queue (DLQ) is full!
So, I fixed the error!!
And now, how can I move dead-letter queue to main queue for consumer process the message again?
@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.
My consumer class looks like:
(tried with with and without the constructor method. Same issue)
import {Consumer} from 'redis-smq';
/**
* Consumer manager
*/
export default class ConsumerQueueManager extends Consumer {
constructor() {
super();
}
/**
* @param {Message} message
* @param {function} cb
*/
consume(message, cb) {
console.debug(message);
cb();
}
}
my application kernel (who starts the consumer)
/**
* Run consumer payment process
*/
async runConsumerQueueProcess() {
if (!process.argv.includes('--process-consumer-queue-disabled')) {
this._logger.info('Starting consumer queue process');
ConsumerQueueManager.queueName = config.queue.name;
const consumerManager = new ConsumerQueueManager();
consumerManager.run();
}
}
got this error:
error: [general] TypeError: Class constructor Consumer cannot be invoked without 'new'
consumer.js
In line 220, its should be message.uuid instead of message.id
this[sLogger].error(Consumer failed to consume message [${message.id}]...
);
so minor :)
Message well send to all running Consumer when the function consume call cb(err) on some Consumer.
Is there bug.
Hi.
Does your app have or will have support for a clustered redis database?
Keep up the good work.
Let's say we have some items which are higher priority (for example, because they are fast to process).
Is there a recommended way to enqueue them at or near the front of the queue?
AbortError: Connection forcefully ended and command aborted. It might have been processed.
at RedisClient.flush_and_error (/data/site/quan-share/node_modules/redis/index.js:357:23)
at RedisClient.end (/data/site/quan-share/node_modules/redis/lib/extendedApi.js:51:14)
at handler (/data/site/quan-share/node_modules/redis-smq/src/scheduler.js:180:24)
at Object.stop (/data/site/quan-share/node_modules/redis-smq/src/scheduler.js:193:20)
at Object.shutdown (/data/site/quan-share/node_modules/redis-smq/src/dispatcher.js:390:27)
at Producer.shutdown (/data/site/quan-share/node_modules/redis-smq/src/producer.js:62:25)
at producer.produce (/data/site/quan-share/utils/log.js:135:16)
at Command.redisClientInstance.lpush [as callback] (/data/site/quan-share/node_modules/redis-smq/src/dispatcher.js:431:25)
at normal_reply (/data/site/quan-share/node_modules/redis/index.js:721:21)
at RedisClient.return_reply (/data/site/quan-share/node_modules/redis/index.js:819:9)
at JavascriptRedisParser.returnReply (/data/site/quan-share/node_modules/redis/index.js:192:18)
at JavascriptRedisParser.execute (/data/site/quan-share/node_modules/redis-parser/lib/parser.js:574:12)
at Socket. (/data/site/quan-share/node_modules/redis/index.js:274:27)
at emitOne (events.js:116:13)
at Socket.emit (events.js:211:7)
at addChunk (_stream_readable.js:263:12)
at readableAddChunk (_stream_readable.js:250:11)
at Socket.Readable.push (_stream_readable.js:208:10)
at TCP.onread (net.js:607:20)
Hello admin,
I had a problem with stats.js on line 78. I think need remove comma at the end of the line 77.
D:\server\chat-server\node_modules\redis-smq\src\stats.js:78
);
^
SyntaxError: Unexpected token )
at createScript (vm.js:53:10)
at Object.runInThisContext (vm.js:95:10)
at Module._compile (module.js:543:28)
at Object.Module._extensions..js (module.js:580:10)
at Module.load (module.js:488:32)
at tryModuleLoad (module.js:447:12)
at Function.Module._load (module.js:439:3)
at Module.require (module.js:498:17)
at require (internal/module.js:20:19)
at Object. (D:\server\chat-server\node_modules\redis-smq\src\consumer.js:8:22)
A declarative, efficient, and flexible JavaScript library for building user interfaces.
๐ Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. ๐๐๐
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google โค๏ธ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.