Task Priority in Celery — Complete Guide
In this tutorial, you will learn about Task Priority in Celery. We cover key concepts, practical examples, and best practices to help you master this topic.
Control task execution order in Celery using priority levels, configure queue priorities, and ensure critical tasks are processed before lower priority work.
What You Learn
You will learn how to configure queue priority support, submit tasks with different priority levels, how workers pick high-priority tasks first, and the limitations of priority with different brokers.
Why It Matters
Without priorities, all tasks are equal. A batch of 10,000 low-priority Background Jobs can delay a critical payment notification by minutes. Priorities ensure that important tasks jump the queue and get processed immediately.
Real-World Use
Doda Browser uses three priority levels: 9 for malware alerts (Process within 1 second), 5 for normal scan results (process within 30 seconds), and 1 for cleanup tasks (process when idle). This ensures threats are always handled first.
Enabling Queue Priorities
Priorities require broker support. Redis supports priority via sorted sets. RabbitMQ supports priority via queue arguments.
from celery import Celery
app = Celery('priority_demo', broker='redis://localhost:6379/0')
# Enable priority support (Redis)
app.conf.update(
task_queue_max_priority=10,
task_default_priority=5,
task_queues={
'default': {
'routing_key': 'default.#',
'queue_arguments': {'x-max-priority': 10},
},
'high': {
'routing_key': 'high.#',
'queue_arguments': {'x-max-priority': 10},
},
},
)
Submitting Tasks with Priority
from celery import Celery
app = Celery('priority_submit', broker='redis://localhost:6379/0')
app.conf.update(
task_queue_max_priority=10,
task_default_priority=5,
)
@app.task
def process_item(item_id, description):
return f"Processed {item_id}: {description}"
# Submit tasks with different priorities
results = []
# Critical - highest priority
r = process_item.apply_async(
args=(1, "Critical failure"),
priority=9
)
results.append((9, r))
# Normal - default priority
r = process_item.apply_async(
args=(2, "Standard update"),
priority=5
)
results.append((5, r))
# Background - lowest priority
r = process_item.apply_async(
args=(3, "Cleanup task"),
priority=1
)
results.append((1, r))
print("Tasks submitted by priority:")
for priority, result in sorted(results, reverse=True):
print(f" Priority {priority}: {result.id[:8]}")
Expected output:
Tasks submitted by priority:
Priority 9: 550e8400...
Priority 5: 6ba7b810...
Priority 1: 6ba7b811...
Priority with RabbitMQ
RabbitMQ requires explicit queue priority configuration:
from celery import Celery
app = Celery('rmq_priority', broker='amqp://guest:guest@localhost:5672//')
app.conf.update(
task_queues={
'priority_queue': {
'routing_key': 'priority.#',
'queue_arguments': {
'x-max-priority': 10,
},
},
},
task_queue_max_priority=10,
task_default_priority=5,
)
@app.task
def prioritized_task(data, level):
return f"[P{level}] {data}"
# RabbitMQ respects priority order within the queue
for priority in range(10, 0, -1):
prioritized_task.apply_async(
args=(f"task_{priority}", priority),
priority=priority
)
print("All priority tasks submitted")
Priority with Multiple Queues
Combine priority with separate queues for maximum control:
from celery import Celery
app = Celery('multi_priority', broker='redis://localhost:6379/0')
app.conf.update(
task_queues={
'critical': {
'routing_key': 'critical.#',
'queue_arguments': {'x-max-priority': 10},
},
'normal': {
'routing_key': 'normal.#',
'queue_arguments': {'x-max-priority': 5},
},
'background': {
'routing_key': 'background.#',
'queue_arguments': {'x-max-priority': 3},
},
},
task_routes={
'tasks.send_alert': {'queue': 'critical'},
'tasks.process_order': {'queue': 'normal'},
'tasks.cleanup_logs': {'queue': 'background'},
},
)
@app.task
def send_alert(user_id, message):
return f"ALERT to {user_id}: {message}"
@app.task
def process_order(order_id):
return f"Order {order_id} processed"
@app.task
def cleanup_logs(days):
return f"Logs older than {days} days cleaned"
Worker configuration:
# Critical worker - always on, low latency
celery -A tasks worker --queues=critical --concurrency=4 --hostname=critical@%h
# Normal worker - standard throughput
celery -A tasks worker --queues=normal --concurrency=8
# Background worker - elastic, can be slow
celery -A tasks worker --queues=background --concurrency=2
Priority and Task Ordering
Demonstrating that high-priority tasks are consumed first:
from celery import Celery
import time
import threading
app = Celery('order_test', broker='redis://localhost:6379/0')
app.conf.update(
task_queue_max_priority=10,
task_default_priority=5,
)
@app.task
def demo_task(name):
print(f"Executing: {name}")
return name
def producer():
# Submit low priority first, then high priority
demo_task.apply_async(args=("low_priority_1",), priority=1)
demo_task.apply_async(args=("low_priority_2",), priority=1)
time.sleep(1) # Give time for workers to pick up
demo_task.apply_async(args=("high_priority_1",), priority=9)
demo_task.apply_async(args=("high_priority_2",), priority=9)
# Start producer in thread
t = threading.Thread(target=producer, daemon=True)
t.start()
time.sleep(5)
# High priority tasks should execute before low priority ones
Priority with Rate Limits
Combine priority with rate limits:
from celery import Celery
app = Celery('rate_priority', broker='redis://localhost:6379/0')
app.conf.update(
task_queue_max_priority=10,
task_default_priority=5,
)
@app.task(rate_limit='10/m')
def prioritized_and_rated(data):
"""Rate-limited task that still respects priority."""
return f"Rated: {data}"
# High priority tasks get processed first within the rate limit
high = prioritized_and_rated.apply_async(args=("urgent",), priority=9)
low = prioritized_and_rated.apply_async(args=("later",), priority=1)
Common Mistakes
1. Not Setting x-max-priority on Queue
Without x-max-priority queue argument, RabbitMQ ignores priority values. Tasks are processed FIFO regardless of priority.
2. Setting Priority Higher Than x-max-priority
Priority values above x-max-priority are clamped. If x-max-priority=5 and you submit with priority=9, it is treated as 5.
3. Expecting Perfect Priority Ordering
Priority is best-effort, not guaranteed. A worker currently processing a low-priority task does not preempt it for a higher-priority task. Priorities only affect which task is picked next.
4. Using Priority Without Separate Queues
If all priorities share one queue, high-priority tasks only jump ahead of waiting tasks. For true isolation, use separate queues per priority.
5. Ignoring Worker Prefetch
If a worker has prefetch=10 and picks up 10 low-priority tasks, it processes all 10 before the worker can pick high-priority tasks. Set low prefetch for priority-sensitive queues.
Practice Questions
1. How do you enable priority support in Celery?
Set task_queue_max_priority=10 and task_default_priority=5. For RabbitMQ, also set x-max-priority queue argument.
2. What is the valid range for priority values?
0-9 when task_queue_max_priority=10. 9 is highest, 0 is lowest. Default is 5.
3. Does Redis support task priorities?
Yes. Redis uses sorted sets internally. Priorities work well with Redis broker.
4. Why might low-priority tasks still execute before high-priority ones?
Workers may have already prefetched low-priority tasks. Set prefetch_count=1 on priority-sensitive queues to minimize this.
Challenge
Design a priority system for a customer support ticket processing system. Tiers: VIP (priority 9, 30-second SLA), Premium (priority 7, 5-minute SLA), Standard (priority 5, 1-hour SLA), Free (priority 1, best-effort). Define queues, priorities, worker configurations, and monitoring thresholds for each tier.
FAQ
Mini Project: Priority Task System
# priority_demo.py
from celery import Celery
import time
import random
import threading
app = Celery('priority_system', broker='redis://localhost:6379/0')
app.conf.update(
task_queue_max_priority=10,
task_default_priority=5,
task_queues={
'default': {
'routing_key': 'default.#',
'queue_arguments': {'x-max-priority': 10},
},
},
)
@app.task(bind=True)
def process_with_priority(self, item_id, level):
processing_time = random.uniform(0.1, 0.5)
time.sleep(processing_time)
result = f"Item {item_id} (P{level}) processed in {processing_time:.2f}s"
print(result)
return result
def batch_submit():
"""Submit tasks with mixed priorities."""
for i in range(20):
priority = random.choice([1, 5, 9])
process_with_priority.apply_async(
args=(i, priority),
priority=priority
)
if __name__ == '__main__':
print("Submitting 20 tasks with mixed priorities (1, 5, 9)...")
t = threading.Thread(target=batch_submit, daemon=True)
t.start()
time.sleep(10)
print("\nExpected execution order: priority 9 tasks first, then 5, then 1")
print("(Workers should pick high-priority tasks before low-priority ones)")
What's Next
Now that you understand task priority, explore periodic tasks with Celery Beat for scheduled task execution, then learn about task result backend for storing and retrieving results.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro