Skip to content

Latest commit

 

History

History
441 lines (332 loc) · 9.11 KB

drivers.md

File metadata and controls

441 lines (332 loc) · 9.11 KB

Queue Drivers

RouteMQ supports two queue drivers: Redis and Database. This guide explains how they work and when to use each.

Overview

Queue drivers handle the storage and retrieval of jobs. The driver you choose affects:

  • Performance - How fast jobs are queued and processed
  • Persistence - Whether jobs survive crashes
  • Infrastructure - What services you need to run
  • Scalability - How well it handles high job volumes

Redis Queue Driver

Fast, in-memory queue backed by Redis.

Features

  • ✅ Very fast - In-memory storage for low latency
  • ✅ High throughput - Handles thousands of jobs/second
  • ✅ Atomic operations - Uses RPOPLPUSH for safe job claiming
  • ✅ Delayed jobs - Efficient sorted sets for scheduling
  • ✅ Scalable - Easy to cluster Redis for more capacity

###Requirements

  • Redis server running (v5.0+)
  • ENABLE_REDIS=true in .env
  • QUEUE_CONNECTION=redis in .env

Configuration

# Enable Redis
ENABLE_REDIS=true
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_DB=0
REDIS_PASSWORD=

# Use Redis for queue
QUEUE_CONNECTION=redis

Data Structures

Redis queue uses different data structures for different purposes:

Structure Purpose Type
routemq:queue:{name} Pending jobs List (FIFO)
routemq:queue:{name}:delayed Delayed jobs Sorted Set (by timestamp)
routemq:queue:{name}:reserved Processing jobs List
routemq:queue:failed:{name} Failed jobs List

How It Works

Pushing a job:

1. Job serialized to JSON
2. RPUSH to routemq:queue:{name}
3. Job ID returned

Popping a job:

1. Check for delayed jobs ready to process
2. RPOPLPUSH from pending to reserved
3. Increment attempts
4. Return job data

Completing a job:

1. LREM from reserved list
2. Job deleted

Failing a job:

1. If attempts < max_tries:
   - LREM from reserved
   - RPUSH back to pending (or delayed)
2. Else:
   - Move to failed list
   - LREM from reserved

Advantages

  • Speed: Sub-millisecond latency
  • Throughput: Handle high job volumes
  • Simple: No complex queries needed
  • Scalable: Easy to add Redis replicas

Disadvantages

  • Volatility: Jobs lost if Redis crashes (unless AOF enabled)
  • Infrastructure: Requires Redis server
  • Memory: All jobs stored in RAM

Best For

  • High-volume applications
  • Real-time job processing
  • When you already use Redis
  • When speed is critical

Database Queue Driver

Persistent queue backed by MySQL.

Features

  • ✅ Persistent - Jobs survive crashes
  • ✅ ACID - Transactional guarantees
  • ✅ Inspectable - Easy to query with SQL
  • ✅ Reliable - No job loss
  • ✅ No extra service - Uses existing MySQL

Requirements

  • MySQL server running (v8.0+)
  • ENABLE_MYSQL=true in .env
  • QUEUE_CONNECTION=database in .env

Configuration

# Enable MySQL
ENABLE_MYSQL=true
DB_HOST=localhost
DB_PORT=3306
DB_NAME=mqtt_framework
DB_USER=root
DB_PASS=your_password

# Use database for queue
QUEUE_CONNECTION=database

Database Tables

queue_jobs:

Column Type Description
id INT Primary key
queue VARCHAR(255) Queue name
payload TEXT Serialized job data
attempts INT Number of attempts
reserved_at DATETIME When job was claimed (NULL if pending)
available_at DATETIME When job becomes available
created_at DATETIME When job was created

queue_failed_jobs:

Column Type Description
id INT Primary key
connection VARCHAR(255) Connection name
queue VARCHAR(255) Queue name
payload TEXT Serialized job data
exception TEXT Exception details
failed_at DATETIME When job failed

How It Works

Pushing a job:

INSERT INTO queue_jobs
(queue, payload, attempts, available_at, created_at)
VALUES (?, ?, 0, ?, NOW())

Popping a job:

-- Use FOR UPDATE SKIP LOCKED for concurrency
SELECT * FROM queue_jobs
WHERE queue = ?
  AND reserved_at IS NULL
  AND available_at <= NOW()
ORDER BY id
LIMIT 1
FOR UPDATE SKIP LOCKED;

-- Mark as reserved
UPDATE queue_jobs
SET reserved_at = NOW(), attempts = attempts + 1
WHERE id = ?

Completing a job:

DELETE FROM queue_jobs WHERE id = ?

Failing a job:

-- If retrying
UPDATE queue_jobs
SET reserved_at = NULL,
    available_at = DATE_ADD(NOW(), INTERVAL ? SECOND)
WHERE id = ?

-- If permanently failed
INSERT INTO queue_failed_jobs
(connection, queue, payload, exception, failed_at)
VALUES (?, ?, ?, ?, NOW());

DELETE FROM queue_jobs WHERE id = ?

Advantages

  • Persistence: Jobs survive crashes
  • Reliability: ACID transactions
  • Visibility: Easy to inspect with SQL
  • Simplicity: No additional infrastructure

Disadvantages

  • Speed: Slower than Redis (disk I/O)
  • Load: Increases database queries
  • Scaling: Harder to scale than Redis

Best For

  • Low to medium job volumes
  • When persistence is critical
  • When you don't want to manage Redis
  • When you need to inspect jobs easily

Comparison

Feature Redis Database
Speed ⭐⭐⭐⭐⭐ Very Fast ⭐⭐⭐ Moderate
Persistence ⭐⭐ Configurable ⭐⭐⭐⭐⭐ Always
Scalability ⭐⭐⭐⭐⭐ Excellent ⭐⭐⭐ Good
Reliability ⭐⭐⭐ Good ⭐⭐⭐⭐⭐ Excellent
Setup ⭐⭐⭐ Need Redis ⭐⭐⭐⭐⭐ Use existing DB
Inspection ⭐⭐⭐ Redis CLI ⭐⭐⭐⭐⭐ SQL queries
Memory ⭐⭐ All in RAM ⭐⭐⭐⭐⭐ On disk

Switching Drivers

You can switch drivers at any time:

# Change in .env
QUEUE_CONNECTION=redis  # or 'database'

Important notes:

  • Jobs in the old driver won't be transferred
  • Complete existing jobs before switching
  • Or manually migrate jobs between drivers

Performance Tips

Redis

# Use a dedicated Redis database
REDIS_DB=1  # Separate from cache

# Enable persistence (optional)
# In redis.conf:
# appendonly yes
# appendfsync everysec

Database

-- Add indexes for performance
CREATE INDEX idx_queue_reserved ON queue_jobs(queue, reserved_at);
CREATE INDEX idx_available ON queue_jobs(available_at);

-- Monitor slow queries
SHOW FULL PROCESSLIST;

Monitoring

Redis

# Connect to Redis
redis-cli

# Check queue sizes
LLEN routemq:queue:default
LLEN routemq:queue:emails
LLEN routemq:queue:reports

# Check delayed jobs
ZCARD routemq:queue:default:delayed

# View pending jobs
LRANGE routemq:queue:default 0 9

# Check memory usage
INFO memory

# Monitor commands
MONITOR

Database

-- Check pending jobs by queue
SELECT queue, COUNT(*) as pending_jobs
FROM queue_jobs
WHERE reserved_at IS NULL
GROUP BY queue;

-- Check processing jobs
SELECT queue, COUNT(*) as processing_jobs
FROM queue_jobs
WHERE reserved_at IS NOT NULL
GROUP BY queue;

-- Check job age
SELECT queue,
       MIN(created_at) as oldest_job,
       MAX(created_at) as newest_job,
       COUNT(*) as total_jobs
FROM queue_jobs
GROUP BY queue;

-- Check failed jobs
SELECT queue, COUNT(*) as failed_jobs
FROM queue_failed_jobs
GROUP BY queue;

-- Find stuck jobs (reserved > 1 hour ago)
SELECT * FROM queue_jobs
WHERE reserved_at < DATE_SUB(NOW(), INTERVAL 1 HOUR);

Troubleshooting

Redis Connection Issues

# Test connection
redis-cli ping

# Check if Redis is running
redis-cli INFO server

# Restart Redis
# macOS: brew services restart redis
# Linux: sudo systemctl restart redis

Database Connection Issues

# Test connection
mysql -h localhost -u root -p -e "SELECT 1"

# Check if MySQL is running
sudo systemctl status mysql

# Check for locks
mysql> SHOW PROCESSLIST;
mysql> SHOW OPEN TABLES WHERE In_use > 0;

Jobs Not Processing

  1. Check driver configuration:

    QUEUE_CONNECTION=redis  # Match your setup

  2. Verify service is running:

    # Redis
redis-cli ping

# MySQL
mysql -h localhost -u root -p -e "SELECT 1"

  3. Check worker connection:

    python main.py --queue-work --connection redis
python main.py --queue-work --connection database

Best Practices

1. Choose Based on Requirements

High volume, speed critical → Redis
Persistence critical → Database
Low volume, simple setup → Database
Already using Redis → Redis

2. Monitor Both

Even if using Redis, keep failed jobs in database:

# RedisQueue already does this
await driver.failed(connection, queue, payload, exception)
# Stores in database if available

3. Regular Cleanup

-- Clean old failed jobs (> 30 days)
DELETE FROM queue_failed_jobs
WHERE failed_at < DATE_SUB(NOW(), INTERVAL 30 DAY);

4. Backup Important Queues

# Redis
redis-cli SAVE
cp /var/lib/redis/dump.rdb /backup/

# MySQL
mysqldump -u root -p mqtt_framework queue_jobs > backup.sql

Next Steps