This module is currently under development and thus not suitable for production yet. You are encouraged to test and evaluate it and report issues if you find any.
- Robust Queue system based on battle hardened architecture.
- Optimized for maxing out Redis performance.
- Non-polling design using blocking connections.
- Simple but powerful api with many advanced features.
- FIFO / LIFO Queues
- Delayed jobs.
- Batches.
- Permisive BSD License
This module implements a realiable queue system as a Redis module. The queue is based on the popular NodeJs modules Bull and BullMQ and it is backwards compatible with both modules.
Having BullMQ as a Redis module has many advantages, firstly it allows easy interoperability with other languages or platforms as long as they have a Redis client.
Secondly, it brings much better performance, squeezing every CPU cycle as much as possible since all code is native.
And thirdly, it is also easier to use and simpler. In the NodeJS versions of the queue we need to perform some bookkeeping activities, such as moving delayed jobs back to the wait queue when they expire, or taking checking if any jobs have been stalled. All this operations are now integrated in the module in a much more efficient way and transparent for the user.
The development of this module is currently sponsored by https://taskforce.sh. Please consider subscribing to help maintaining this library.
Download a release and install it in your redis instance using the load module command:
> MODULE LOAD /path/to/libbullmq.soIt is also possible to load the module automatically when starting redis by specifying the following directive in the redis.conf file:
loadmodule /path/to/libbullmq.soThis module follows SemVer and releases are automatically released as long as there is at least one fix.
In the context of BullMQ we talk about jobs instead of messages. The different is subtle, but Job could be seen as a super set of a message, where a message is just some information passed from A to B, a Job includes the notion of some work to be done when the message is received. Other than that, the semantics of the queue should be identical for jobs or messages.
QADD queueName { jobName data opts [ID jobID] [DELAY millis] [LIFO] }+
This command lets you add one or more jobs to the queue in one call.
Arguments:
- queueName: the name of the queue name. For interoperability with existing libraries you should
use a name with a prefix, like: "bull:my-queue-name"
- jobName: a name for the job. Useful for clasifying the jobs in UIs, can also be useful
when processing jobs if you want to perform different tasks depending on the job name.
- data: a string containinig the data. For interoperability with existing tools you should store
data as a JSON string, but you can use any data you like including binary data.
- opts: options useful for building high level functionality om top of existing BullMQ commands,
for example it can be used to store the maximum amount of times a job should be retried in the case
of failure.
- [ID jobID]: instead of letting BullMQ generate a jobId you can specify your own. Important to notice
is that BullMQ will ignore jobs with the same ID (that are still in the queue). This property can be
exploited to avoid duplication of jobs in some scenarios.
- [DELAY millis]: milliseconds to delay this job. Instead of putting the job in the wait list, it is
placed in the delayed set. As soon as the job has waited the specified time it is returned in the next
call to QNEXT. Note, delayed jobs will skip the wait queue even if there are other non-delayed jobs
waiting.
- [LIFO]: use this arg if you want the queue to return the jobs in Last In First Out order instead of
the default (First In First out) order.
QNEXT queueName token lockTTL [BATCH numJobs] [timeout]
This command optional blocks if no jobs are available at the moment and a timeout is passed as argument.
Arguments:
- queueName: the name of the queue.
- token: a unique token used for locking the job.
- lockTTL: time to live for the lock. The job will be automatically moved back to wait status after
this time has passed. (Unless the job has completed or failed).
- [BATCH numJobs] number of jobs to get at once in a batch.
Batch has 2 modes of operation
- standard batch, returns as much jobs as possible (less or equal than numJobs) without waiting.
- batch with timeout, blocks until numJobs are available or timeout returning as many jobs as possible if
it timesout.
Regarding batches there are some important considerations. All the jobs returned will be locked with lockTTL. You need to make sure you are done with these jobs before the lockTTL has expired or renew the lock time calling QRELOCK. If you fail to do so, other callers to QNEXT will receive the jobs that have expired.
QDONE queue token { jobId [RESULT result] | [ERROR err] }+ [TRIM numJobs] [NEXT loockTTL numJobs timeout]
QRELOCK queue token lockTTL { jobId }+
This is used in order mark that the worker is still working on the job so that other workers calling QNEXT do not get the same job again.
QINFO [LIMIT] [JOBS]
Get info about different aspects of the queue.
QPAUSE queue
Pauses a queue, it is possible to add and complete jobs but QNEXT will not return any job until it is resumed again.
QRESUME queue
Resumes a paused queue.
QISPAUSED queue
Checks of queue is currently paused or running, returns "PAUSED" or "RUNNING".
Get jobs (returns also last event ID)
QDEL queue jobId
Removes one job from the queue.
Update a job. It is possible to update progress, data, opts or even adding a new log message.
QDISPOSE queue
Removes all data related to a Queue.
QCLEAN [WAIT] [ACTIVE] [DELAYED] [COMPLETED] [FAILED]
Removes jobs from the given states.
Queues produce events that are pushed to redis streams. This is very convenient since it allows clients to consume events at their own pace avoiding missing any event.
BullMQ is really fast. On any modern hardware you should expect easily 50k jobs/sec. The bottlenecks will likely be on the network IO and job processing itself rather than on BullMQ. If you use batching you can increase this numbers easily an order or magnitude.
We describe some common patterns, for example processing jobs, batch processing, progress, event listening, etc, etc.
In this section we give an overview of BullMQ architecture, how it is built on top of redis, which redis datatructure it uses an so forth.
BullMQ is based on a "wait" list, "active" list and a "complete" and "failed" sets. Jobs are added to the wait list and as they are processed moved to the "active" list. When the job/message is finally processed it is moved to the completed or failed sets depending if the job succeeded respectively failed.
In reality in this list we only store the job id, the job data is stored in a separate hash map, so when we say that we are moving a job from one list to another we mean that we are just moving the job id.
On top of this basic architecture, all other features are built. Lets go one by one from simplest to more complex
As jobs are moved to the active list they are also locked. The QNEXT command requires a lock token as well as a lock TTL. The token is used to specify the "owner" of the job that is being processed. It is recommended to use a unique uuid for the token. This lock is used for a couple of things, first to mark who owns the job but more importantly to detect if the worker processing the job has crashed or hanged. This is because the worker has just lockTTL time to process the job before BullMQ will move the job back to the wait list or, if the number of stalled counts is larger than a limit, moved to the failed set. However a worker that knows the lock token can call QRELOCK to renew the lock for more time, telling BullMQ that the worker is still alive and processing the job.
When the job is moved from active to complete of failed using QDONE, the lock is also released. As you can see QDONE requires the lock token, so no one other than the worker that actually started processing the job can mark a job as done.
Copyright (c) 2021, Taskforce.sh Inc. All rights reserved.
Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met:
- Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer.
- Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution.
- All advertising materials mentioning features or use of this software must display the following acknowledgement: This product includes software developed by the Taskforce.sh Inc..
- Neither the name of the Taskforce.sh Inc. nor the names of its contributors may be used to endorse or promote products derived from this software without specific prior written permission.
THIS SOFTWARE IS PROVIDED BY Taskforce.sh Inc. ''AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL Taskforce.sh Inc. BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
(c) 2021 Taskforce.sh Inc.
The source code and all the texts in this repository excluding issues reported by third party belongs and are copyrighted by Taskforce.sh Inc.