repository·master·Indexed 19 days ago
https://github.com/pricingassistant/mrqMRQ is an opinionated, distributed task queue for Python that uses MongoDB for job and log storage and Redis for queue management. It provides tools for asynchronous task execution via mrq-worker, synchronous execution via mrq-run, and a monitoring interface via mrq-dashboard. The library supports scheduled tasks, timed set queues, and high-throughput IO-bound processing using Gevent greenlets.
MRQ provides several features for managing distributed tasks:
objgraph.A Worker is the unit of processing in MRQ responsible for dequeuing jobs and executing them.
Key characteristics:
When using timed set queues, the execution of tasks is tied to their scheduled timestamp.
Important Timing Note: If you enqueue tasks for the future but delay starting your worker, the worker will immediately execute any tasks whose scheduled execution time has already passed.
For example, if you enqueue tasks every 10 seconds but wait 20 seconds before starting the worker, the first two tasks will be executed immediately because their scheduled time is in the past. Subsequent tasks will then follow the intended interval relative to their scheduled times.
Raw queues prioritize performance by storing only serialized task parameters in Redis. Tasks are only inserted into MongoDB after being dequeued by a worker. This reduces visibility for individual queued jobs but increases throughput.
There are four types of raw queues determined by their name suffix:
_raw: Simplest type, stored in a Redis LIST._set: Stored in a Redis SET. Supports "unique" tasks (only one instance of a specific task/parameter pair can be queued at a time)._sorted_set: Stored in a Redis ZSET. Allows ordering/re-ordering tasks. Like _set, task parameters are unique._timed_set: A special _sorted_set where tasks are sorted by a UNIX timestamp, allowing you to schedule tasks for a specific time in the future.MRQ distinguishes between the definition of work and its execution:
mrq.task.Task) that wraps a unit of processing. It defines what to do.A Task can invoke other tasks either synchronously or asynchronously by queuing them as new Jobs.
MRQ includes a built-in scheduler that allows you to run tasks at regular intervals (e.g., every X units of time), similar to how crontab works.
If you need to schedule a job to run at a specific, precise time in the future rather than on a recurring interval, use timed_set available via raw queues.
Regular queues store tasks directly in MongoDB. By default, they follow a FIFO (First-In-First-Out) pattern. You can transform a regular queue into a LIFO (Last-In-First-Out) 'pile' by appending _reverse to the queue name when starting a worker.
# To dequeue the last jobs added to the queue "default" (LIFO)
$ mrq-worker default_reverseJobs transition through various statuses. Understanding these is critical for monitoring and error handling:
Standard Lifecycle:
queued: Created and waiting for a Worker.started: A Worker has begun execution.success: Execution completed successfully.Error and Interruption States:
failed: A Python Exception was raised during execution.cancel: The job was cancelled (usually via the Dashboard). Note: Cancelling a started job does not interrupt the currently running code.abort: The job was stopped via abort_current_job(). Used for unrecoverable errors you want to log without retrying.interrupt: The worker process received a SIGTERM or two SIGINTs. (Note: SIGKILL or power loss will leave the job in started state).timeout: The job exceeded its configured timeout.retry: task.retry() was called to schedule a retry.maxretries: The task reached its maximum allowed retry attempts (default is 3).Choosing a queue type involves balancing performance, dashboard visibility, and job safety.
| Queue type | Regular | Raw | Raw with no_storage config |
|---|---|---|---|
| Storage for queued jobs | MongoDB | Redis | Redis |
| Storage for started & success jobs | MongoDB | MongoDB | None |
| Performance | + | ++ | +++ |
| Visibility in the dashboard | Full | After start | Job counts & failed jobs |
| Safety | +++ | ++ | + |
MRQ does not support Graphite out of the box, but you can integrate it by implementing a METRIC_HOOK in your mrq-config file. This hook is called by MRQ whenever a metric is recorded.
To use Graphite, you should install the graphiteudp package via pip, initialize a GraphiteUDPClient, and then define the METRIC_HOOK function to send the metric name and increment value to your Graphite instance. You can use logic within the hook (such as a whitelist) to filter which metrics are sent to avoid overwhelming your monitoring system.
import graphiteudp # Install this via pip
# Initialize the Graphite UDP Client
_graphite_client = graphiteudp.GraphiteUDPClient(host, port, prefix, debug=False)
_graphite_client.init()
def METRIC_HOOK(name, incr=1, **kwargs):
# You can use this to avoid sending too many different metrics
whitelisted_metrics = ["queues.all.", "queues.default.", "jobs."]
if any([name.startswith(m) for m in whitelisted_metrics]):
_graphite_client.send(name, incr)