Job Priorities in Background Processing
In this tutorial, you will learn about Job Priorities in Background Processing. We cover key concepts, practical examples, and best practices to help you master this topic.
Control background job execution order with priorities, configure priority queues, and ensure critical jobs are processed before lower-priority work.
What You Learn
You will learn how priority queues work, implement priority in different queue systems, and design priority strategies for production applications.
Why It Matters
Without priorities, all jobs are equal. A flood of low-priority cleanup jobs can delay critical payment processing. Priorities ensure that important jobs skip the queue.
Real-World Use
DodaTech uses three priority levels: critical (alerts, payments), normal (email, notifications), and low (cleanup, reporting). Critical jobs are always processed first regardless of queue depth.
Priority Queue Concepts
flowchart LR
P[Producer] --> Q[Priority Queue]
Q -->|Priority 9| H[High Worker]
Q -->|Priority 5| N[Normal Worker]
Q -->|Priority 1| L[Low Worker]
H -->|Always first| P1[Alerts]
N -->|When available| P2[Email]
L -->|When idle| P3[Cleanup]
Priority with Redis Sorted Sets
import redis
import json
import time
r = redis.Redis()
class PriorityQueue:
def __init__(self, name='pq'):
self.name = name
def enqueue(self, job, priority=5):
r.zadd(self.name, {json.dumps(job): -priority})
def dequeue(self):
results = r.zpopmin(self.name, 1)
if results:
return json.loads(results[0][0])
return None
def size(self):
return r.zcard(self.name)
pq = PriorityQueue()
pq.enqueue({'task': 'cleanup'}, priority=1)
pq.enqueue({'task': 'alert'}, priority=9)
pq.enqueue({'task': 'report'}, priority=5)
while pq.size():
job = pq.dequeue()
print(f"Dequeued: {job['task']}")
Expected output:
Dequeued: alert
Dequeued: report
Dequeued: cleanup
Priority with Multiple Queues
import redis
import json
import threading
import time
r = redis.Redis()
class MultiPriorityQueue:
def __init__(self):
self.queues = {
'critical': [],
'high': [],
'default': [],
'low': [],
}
def enqueue(self, job, priority='default'):
r.lpush(f"queue:{priority}", json.dumps(job))
def dequeue(self):
for priority in ['critical', 'high', 'default', 'low']:
data = r.rpop(f"queue:{priority}")
if data:
return json.loads(data), priority
return None, None
mpq = MultiPriorityQueue()
mpq.enqueue({'task': 'backup'}, 'low')
mpq.enqueue({'task': 'payment'}, 'critical')
mpq.enqueue({'task': 'email'}, 'default')
for _ in range(3):
job, queue = mpq.dequeue()
print(f"From '{queue}': {job['task']}")
Expected output:
From 'critical': payment
From 'default': email
From 'low': backup
Priority-Based Worker
import redis
import json
import time
import threading
r = redis.Redis()
class PriorityWorker:
def __init__(self):
self.handlers = {}
self.running = True
self.queues = ['critical', 'high', 'default', 'low']
def task(self, name):
def decorator(func):
self.handlers[name] = func
return func
return decorator
def start(self):
while self.running:
processed = False
for queue in self.queues:
data = r.rpop(f"queue:{queue}")
if data:
job = json.loads(data)
handler = self.handlers.get(job['task'])
if handler:
print(f"[{queue.upper()}] Processing: {job['task']}")
handler(**job.get('data', {}))
processed = True
break
if not processed:
time.sleep(0.5)
def stop(self):
self.running = False
worker = PriorityWorker()
@worker.task('alert')
def alert(msg):
print(f" ALERT: {msg}")
@worker.task('email')
def email(to):
print(f" Sending email to {to}")
@worker.task('cleanup')
def cleanup():
print(f" Running cleanup")
r.lpush('queue:critical', json.dumps({'task': 'alert', 'data': {'msg': 'Server down'}}))
r.lpush('queue:low', json.dumps({'task': 'cleanup'}))
r.lpush('queue:default', json.dumps({'task': 'email', 'data': {'to': 'user@example.com'}}))
t = threading.Thread(target=worker.start, daemon=True)
t.start()
time.sleep(2)
worker.stop()
Expected output:
[CRITICAL] Processing: alert
ALERT: Server down
[DEFAULT] Processing: email
Sending email to user@example.com
[LOW] Processing: cleanup
Running cleanup
Priority in Bull (Node.js)
const Queue = require('bull');
const queue = new Queue('priorities', 'redis://127.0.0.1:6379');
queue.add({ task: 'cleanup' }, { priority: 10 });
queue.add({ task: 'report' }, { priority: 5 });
queue.add({ task: 'critical_alert' }, { priority: 1 });
queue.process(async (job) => {
console.log(`Processing: ${job.data.task} (priority ${job.opts.priority})`);
});
// Higher priority (lower number) jobs process first
Priority in Sidekiq (Ruby)
class PriorityWorker
include Sidekiq::Worker
sidekiq_options priority: 10 # Lower = higher priority
end
class CriticalWorker
include Sidekiq::Worker
sidekiq_options priority: 1
end
# Or set at enqueue time
PriorityWorker.set(priority: 1).perform_async(data) # High priority
PriorityWorker.set(priority: 10).perform_async(data) # Low priority
Common Mistakes
1. Using Too Many Priority Levels
Three levels (critical, normal, low) are sufficient. More levels create confusion and minimal practical benefit.
2. Not Isolating Critical Jobs
High-priority jobs still wait behind normal jobs if sharing a single worker. Isolate critical jobs to dedicated workers.
3. Ignoring Starvation
Low-priority jobs may never execute if critical jobs keep arriving. Use aging: increase the priority of waiting jobs over time.
4. Priority Without Queue Configuration
Some backends require explicit priority configuration (Redis sorted sets, RabbitMQ x-max-priority). Default FIFO ignores priority.
5. Not Monitoring Priority Distribution
Track how many jobs of each priority are processed. If low-priority jobs never get processed, your priority system is broken.
Practice Questions
1. How do priorities work in job queues?
Higher-priority jobs are dequeued before lower-priority ones. The implementation depends on the backend (sorted sets, separate queues, priority fields).
2. What is priority starvation?
Low-priority jobs may never execute if higher-priority jobs keep arriving. Aging mechanisms increase priority of waiting jobs to prevent this.
3. How many priority levels should you use?
3-5 levels maximum: critical, high, normal, low, background. More levels add complexity without significant benefit.
4. Should critical jobs share workers with normal jobs?
No. Give critical jobs dedicated workers to ensure they are never blocked by normal jobs.
Challenge
Design a priority system for a hospital notification system: emergency alerts (critical, dedicated workers, 10-second SLA), appointment reminders (normal, shared workers, 1-hour SLA), billing notifications (low, best-effort SLA), and report generation (background, overnight only).
FAQ
Mini Project: Priority System
import redis
import json
import time
import threading
import random
r = redis.Redis()
priorities = {
'critical': 4,
'high': 3,
'normal': 2,
'low': 1,
}
class PriorityProducer:
def enqueue(self, queue_name, job):
job['priority'] = priorities.get(queue_name, 2)
job['enqueued_at'] = time.time()
r.lpush(f"queue:{queue_name}", json.dumps(job))
class PriorityWorker:
def __init__(self):
self.running = True
def start(self):
while self.running:
for pname in ['critical', 'high', 'normal', 'low']:
data = r.rpop(f"queue:{pname}")
if data:
job = json.loads(data)
self.process(job, pname)
break
else:
time.sleep(0.2)
def process(self, job, queue_name):
print(f"[{queue_name.upper()}] {job.get('task', 'unknown')}")
time.sleep(random.uniform(0.1, 0.5))
def stop(self):
self.running = False
producer = PriorityProducer()
producer.enqueue('low', {'task': 'cleanup'})
producer.enqueue('critical', {'task': 'server_failure_alert'})
producer.enqueue('normal', {'task': 'send_email'})
producer.enqueue('high', {'task': 'payment_processing'})
worker = PriorityWorker()
t = threading.Thread(target=worker.start, daemon=True)
t.start()
time.sleep(3)
worker.stop()
Expected output:
[CRITICAL] server_failure_alert
[HIGH] payment_processing
[NORMAL] send_email
[LOW] cleanup
What's Next
Now that you understand priorities, explore job retries and backoff for handling failures, then learn about job failure handling patterns.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro