Celery Task Queues — Complete Guide
In this tutorial, you will learn about Celery Task Queues. We cover key concepts, practical examples, and best practices to help you master this topic.
Organize Celery tasks into named queues for separation of concerns, control worker queue consumption, and manage queue lifecycle in production.
What You Learn
You will learn how to define and manage multiple task queues, configure workers to consume from specific queues, use queue priorities, and manage queue backlogs.
Why It Matters
A single queue for all tasks creates a bottleneck. A misbehaving task type can block all others. Separate queues isolate different workloads, ensure critical tasks get processed promptly, and allow independent scaling per task type.
Real-World Use
DodaTech uses three queues: critical (immediate processing, always-on workers), default (normal tasks, elastic workers), and batch (overnight jobs, spot instances). Each queue has independent scaling and monitoring.
Defining Multiple Queues
from celery import Celery
app = Celery('queues', broker='redis://localhost:6379/0')
app.conf.update(
task_queues={
'default': {
'routing_key': 'default.#',
'exchange': 'default',
'exchange_type': 'topic',
},
'high': {
'routing_key': 'high.#',
'exchange': 'high',
'exchange_type': 'topic',
},
'low': {
'routing_key': 'low.#',
'exchange': 'low',
'exchange_type': 'topic',
},
},
task_default_queue='default',
task_default_routing_key='default.task',
)
@app.task(queue='high')
def urgent_task(data):
return f"Urgent: {data}"
@app.task(queue='default')
def normal_task(data):
return f"Normal: {data}"
@app.task(queue='low')
def background_task(data):
return f"Background: {data}"
Queue Configuration with Options
from celery import Celery
app = Celery('queue_opts', broker='redis://localhost:6379/0')
app.conf.update(
task_queues={
'default': {
'routing_key': 'default.#',
'queue_arguments': {
'x-max-priority': 10,
},
},
'delayed': {
'routing_key': 'delayed.#',
'queue_arguments': {
'x-message-ttl': 86400000, # 24 hours in ms
'x-max-length': 10000,
},
},
'dead_letter': {
'routing_key': 'dead.#',
'queue_arguments': {
'x-dead-letter-exchange': 'dlx',
'x-dead-letter-routing-key': 'dead.letter',
},
},
}
)
Worker Queue Assignment
# Worker for specific queues
celery -A tasks worker --queues=high --concurrency=4 --hostname=fast-worker@%h
# Worker for multiple queues
celery -A tasks worker --queues=default,low --concurrency=8
# Worker for all queues
celery -A tasks worker --queues=high,default,low --concurrency=12
# Worker with queue weights (Celery 5.x)
celery -A tasks worker --queues=high:4,default:2,low:1
Queue weights control how many tasks from each queue the worker picks:
# Worker with weights: high gets 4 tasks for every 2 default and 1 low
# This ensures high-priority tasks are processed faster
Dynamic Queue Creation
Create queues at runtime for dynamic workloads:
from celery import Celery
from kombu import Queue, Exchange
app = Celery('dynamic', broker='redis://localhost:6379/0')
def create_dynamic_queue(tenant_id):
"""Create a queue for a specific tenant."""
exchange = Exchange(f'tenant_{tenant_id}', type='direct')
queue = Queue(
f'tenant_{tenant_id}',
exchange=exchange,
routing_key=f'tenant.{tenant_id}',
queue_arguments={'x-max-length': 10000}
)
return queue
@app.task
def process_tenant_task(tenant_id, data):
queue = create_dynamic_queue(tenant_id)
print(f"Processing task for tenant {tenant_id} on queue {queue.name}")
return f"Tenant {tenant_id}: {data}"
Queue Monitoring
from celery import Celery
import redis
app = Celery('monitor', broker='redis://localhost:6379/0')
def get_queue_lengths(app):
"""Get the length of all Celery queues."""
client = redis.Redis.from_url(app.conf.broker_url)
queue_lengths = {}
for queue_name, queue_config in app.conf.task_queues.items():
queue_key = queue_name
length = client.llen(queue_key)
queue_lengths[queue_name] = length
return queue_lengths
@app.task
def sample_task(duration):
import time
time.sleep(duration)
return f"Slept for {duration}s"
# Usage
lengths = get_queue_lengths(app)
print("Queue lengths:")
for name, length in lengths.items():
status = "OK" if length < 100 else "WARNING" if length < 1000 else "CRITICAL"
print(f" {name}: {length} tasks [{status}]")
Expected output:
Queue lengths:
default: 0 tasks [OK]
high: 5 tasks [OK]
low: 150 tasks [WARNING]
Queue Purging
from celery import Celery
app = Celery('purge', broker='redis://localhost:6379/0')
# Purge all queues
def purge_all_queues():
for queue_name in app.conf.task_queues:
purged = app.control.purge(queue=queue_name)
print(f"Purged {queue_name}: {purged} messages")
# Purge a specific queue
def purge_queue(queue_name):
purged = app.control.purge(queue=queue_name)
print(f"Purged {queue_name}: {purged} messages")
# Purge all
@app.task
def reset_queues():
total = app.control.purge()
print(f"Purged {total} messages from all queues")
return total
Common Mistakes
1. Using Too Many Queues
Each queue adds overhead. Start with 3-5 queues (critical, default, batch, dead_letter). Add more only when monitoring shows isolation is necessary.
2. Not Setting Queue Limits
Without x-max-length or x-message-ttl, queues can grow unbounded. A failing consumer causes millions of messages to accumulate, exhausting broker memory.
3. Workers Consuming Wrong Queues
A worker consuming from batch queue processes CPU-intensive tasks alongside latency-sensitive high queue tasks. Dedicate separate worker pools per queue type.
4. Forgetting Queue Binding with RabbitMQ
When using RabbitMQ, the exchange must be bound to the queue. Celery handles this automatically if queues are defined in task_queues, but manually created exchanges need explicit binding.
5. Not Monitoring Queue Depth
Queue depth is the most important Celery metric. Set up alerts for queue depth thresholds. A growing queue indicates a bottleneck or consumer failure.
Practice Questions
1. How do you create a queue with a maximum length?
Use queue_arguments={'x-max-length': 10000} in the queue definition. This prevents the queue from growing indefinitely.
2. How do you assign a task to a specific queue?
Set queue='queue_name' in the @app.task decorator, or pass queue='name' in apply_async() when calling the task.
3. What is queue weighting in workers?
Queue weights control how many tasks a worker picks from each queue. --queues=high:4,default:2 means for every 4 high tasks, pick 2 default tasks.
4. How do you purge a queue?
Use app.control.purge(queue='name') or the celery command celery -A app purge -Q queue_name.
Challenge
Design a queue Strategy for a SaaS platform with 3 tiers (free, pro, enterprise). Free tier tasks go to a low-priority queue with Rate Limiting. Pro gets a dedicated queue. Enterprise gets highest priority with 24/7 worker coverage. Implement queue isolation and worker sizing per tier.
FAQ
Mini Project: Queue Management System
# queue_manager.py
from celery import Celery
import redis
import json
import time
app = Celery('queue_mgmt', broker='redis://localhost:6379/0')
app.conf.update(
task_queues={
'critical': {
'routing_key': 'critical.#',
'queue_arguments': {'x-max-priority': 10, 'x-max-length': 10000},
},
'default': {
'routing_key': 'default.#',
'queue_arguments': {'x-max-length': 50000},
},
'batch': {
'routing_key': 'batch.#',
'queue_arguments': {'x-max-length': 100000, 'x-message-ttl': 604800000},
},
'dead_letter': {
'routing_key': 'dead.#',
'queue_arguments': {'x-max-length': 100000},
},
},
task_default_queue='default',
)
def monitor_queues():
client = redis.Redis.from_url(app.conf.broker_url)
while True:
print(f"\n--- Queue Status at {time.strftime('%H:%M:%S')} ---")
for queue_name in app.conf.task_queues:
length = client.llen(queue_name)
status = 'OK' if length < 100 else 'WARN' if length < 1000 else 'CRIT'
print(f" {queue_name:15s} {length:6d} tasks [{status}]")
time.sleep(5)
@app.task(queue='critical')
def critical_task(data):
return f"CRITICAL: {data}"
@app.task(queue='default')
def default_task(data):
return f"DEFAULT: {data}"
@app.task(queue='batch')
def batch_task(data):
return f"BATCH: {data}"
if __name__ == '__main__':
print("Queue Management Demo:")
print("Queues configured:", list(app.conf.task_queues.keys()))
print("\nStart workers:")
print(" celery -A queue_manager worker --queues=critical --concurrency=2")
print(" celery -A queue_manager worker --queues=default --concurrency=4")
print(" celery -A queue_manager worker --queues=batch --concurrency=8")
Expected output:
Queue Management Demo:
Queues configured: ['critical', 'default', 'batch', 'dead_letter']
Start workers:
celery -A queue_manager worker --queues=critical --concurrency=2
celery -A queue_manager worker --queues=default --concurrency=4
celery -A queue_manager worker --queues=batch --concurrency=8
What's Next
Now that you understand task queues, learn about task priority for controlling execution order within queues, then explore periodic tasks with Celery Beat for scheduled execution.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro