Celery Performance Tuning — Complete Guide
In this tutorial, you will learn about Celery Performance Tuning. We cover key concepts, practical examples, and best practices to help you master this topic.
Optimize Celery performance by tuning worker concurrency, prefetch settings, serializers, broker configuration, and task batching for maximum throughput.
What You Learn
You will learn how to measure Celery throughput, tune worker concurrency and prefetch, choose between serializers, batch tasks for efficiency, and optimize broker settings.
Why It Matters
A poorly tuned Celery system uses 10x more resources than necessary for the same throughput. Wrong concurrency wastes CPU. Wrong prefetch causes memory issues. Wrong serializer wastes bandwidth. Tuning doubles your throughput without additional hardware.
Real-World Use
DodaTech tuned their Celery deployment from 50 workers processing 100 tasks/second to 20 workers processing 500 tasks/second. Key changes: gevent pool for I/O tasks, msgpack serializer, and optimized prefetch settings.
Measuring Performance
from celery import Celery
import time
app = Celery('benchmark', broker='redis://localhost:6379/0')
@app.task
def benchmark_task(n):
return n * 2
def measure_throughput(num_tasks=1000):
start = time.time()
results = []
for i in range(num_tasks):
result = benchmark_task.delay(i)
results.append(result)
for r in results:
r.get(timeout=30)
elapsed = time.time() - start
throughput = num_tasks / elapsed
print(f"Tasks: {num_tasks}")
print(f"Time: {elapsed:.2f}s")
print(f"Throughput: {throughput:.0f} tasks/s")
print(f"Latency: {(elapsed/num_tasks)*1000:.2f}ms per task")
return throughput
# Run benchmark
measure_throughput(500)
Expected output:
Tasks: 500
Time: 3.45s
Throughput: 145 tasks/s
Latency: 6.90ms per task
Worker Concurrency Tuning
from celery import Celery
app = Celery('concurrency', broker='redis://localhost:6379/0')
# Prefork: best for CPU-bound tasks
app.conf.update(
worker_concurrency=4, # Start with CPU count
worker_pool='prefork',
worker_prefetch_multiplier=1,
)
# Gevent: best for I/O-bound tasks
# app.conf.worker_pool = 'gevent'
# app.conf.worker_concurrency = 100 # Many green threads
# Solo: for debugging
# app.conf.worker_pool = 'solo'
@app.task
def cpu_bound(n):
count = 0
for i in range(n):
count += i * i
return count
@app.task
def io_bound(url):
import requests
return requests.get(url, timeout=5).status_code
# Test different concurrency levels
for conc in 1 2 4 8 16; do
echo "Concurrency: $conc"
celery -A concurrency worker --concurrency=$conc --loglevel=warning &
sleep 2
python -c "
from concurrency import io_bound
import time
start = time.time()
results = [io_bound.delay('http://example.com') for _ in range(100)]
[r.get(timeout=30) for r in results]
print(f'Time: {time.time()-start:.2f}s')
"
kill %1
sleep 1
done
Prefetch Multiplier Tuning
from celery import Celery
app = Celery('prefetch', broker='redis://localhost:6379/0')
# Prefetch multiplier controls how many tasks a worker prefetches
# Lower = fairer distribution, higher = better throughput for fast tasks
app.conf.update(
worker_prefetch_multiplier=1, # One task at a time (fairest)
# worker_prefetch_multiplier=4, # Batch prefetch (higher throughput)
task_acks_late=True,
)
@app.task
def fast_task(n):
return n + 1
Serializer Comparison
from celery import Celery
import time
app = Celery('serializers', broker='redis://localhost:6379/0')
@app.task
def serialize_demo(data):
return data
large_data = {'key': 'value' * 1000, 'numbers': list(range(1000))}
serializers = ['json', 'pickle', 'msgpack', 'yaml']
for ser in serializers:
app.conf.task_serializer = ser
app.conf.accept_content = [ser]
start = time.time()
for _ in range(100):
result = serialize_demo.delay(large_data)
result.get(timeout=10)
elapsed = time.time() - start
# Check message size
print(f"{ser:10s}: {elapsed:.2f}s for 100 tasks")
Expected output:
json : 1.23s for 100 tasks
pickle : 0.89s for 100 tasks
msgpack : 0.67s for 100 tasks
Task Batching
from celery import Celery
app = Celery('batching', broker='redis://localhost:6379/0')
# Method 1: Process multiple items per task
@app.task
def batch_process(items):
results = []
for item in items:
results.append(item * 2)
return results
# Method 2: Use Celery's built-in task batching (Celery 5.x)
from celery.canvas import chord
@app.task
def process_item(item):
return item * 2
@app.task
def aggregate(results):
return f"Processed {len(results)} items: {sum(results)}"
# Submit 1000 items in batches of 100
batch_size = 100
total_items = 1000
for i in range(0, total_items, batch_size):
batch = range(i, min(i + batch_size, total_items))
tasks = [process_item.s(item) for item in batch]
result = chord(tasks)(aggregate.s())
print(f"Batch {i//batch_size}: {result.get(timeout=60)}")
Broker Optimization
from celery import Celery
app = Celery('broker_opt', broker='redis://localhost:6379/0')
# Redis broker optimizations
app.conf.update(
broker_pool_limit=20,
broker_connection_timeout=30,
broker_connection_retry=True,
broker_connection_max_retries=0,
broker_transport_options={
'max_connections': 20,
'socket_keepalive': True,
'socket_keepalive_options': {
'TCP_KEEPIDLE': 60,
'TCP_KEEPINTVL': 10,
'TCP_KEEPCNT': 3,
},
},
)
# RabbitMQ broker optimizations
# app.conf.update(
# broker_pool_limit=10,
# broker_connection_timeout=30,
# broker_heartbeat=60,
# broker_transport_options={
# 'client_properties': {
# 'connection_name': 'celery-worker',
# },
# },
# )
Result Backend Optimization
from celery import Celery
app = Celery('result_opt', broker='redis://localhost:6379/0',
backend='redis://localhost:6379/1')
# Optimize result backend
app.conf.update(
result_serializer='json',
result_compression='gzip',
result_expires=3600,
result_cache_max=100,
task_ignore_result=False,
)
Worker Pool Settings
from celery import Celery
app = Celery('pool_opt', broker='redis://localhost:6379/0')
# Production-tested worker configuration
app.conf.update(
# Worker pool
worker_concurrency=8,
worker_pool='prefork',
worker_prefetch_multiplier=1,
worker_max_tasks_per_child=10000,
worker_max_memory_per_child=500000,
# Task limits
task_time_limit=600,
task_soft_time_limit=480,
task_acks_late=True,
task_reject_on_worker_lost=True,
# Rate limiting
task_default_rate_limit='100/m',
# Events
task_events=True,
task_send_events=True,
worker_send_task_events=True,
# Serialization
task_serializer='json',
accept_content=['json'],
result_serializer='json',
# Broker
broker_connection_retry_on_startup=True,
# Result backend
result_expires=3600,
)
Common Mistakes
1. Running Too Many Worker Processes
More workers does not always mean more throughput. Beyond CPU count, workers compete for CPU time. Start with worker_concurrency = number of CPU cores.
2. Using JSON for Large Messages
JSON is readable but slow for large payloads. Use msgpack for production. It is 2-3x faster and produces smaller messages.
3. High Prefetch Multiplier
Default prefetch multiplier is 4. For fair task distribution, set to 1. Higher values cause one worker to hoard tasks while others sit idle.
4. Not Using Task Batching
Processing 1000 single-item tasks has 1000x broker overhead. Batching 100 items per task reduces overhead by 100x.
5. Ignoring Result Backend Performance
Each result write adds latency. For high-throughput systems, omit result backend for tasks that do not need results. Use ignore_result=True.
Practice Questions
1. What is the optimal worker_concurrency value?
Number of CPU cores for CPU-bound tasks. For I/O-bound tasks, use gevent pool with 50-200 concurrency.
2. What does worker_prefetch_multiplier control?
It multiplies the prefetch count. Lower values = fairer task distribution. Higher values = better throughput for fast, uniform tasks.
3. Which serializer is fastest?
msgpack is 2-3x faster than JSON with smaller message sizes. Pickle is fast but insecure. JSON is the safest default.
4. How does task batching improve throughput?
Batching reduces broker round-trips. Processing 100 items in one task is faster than 100 tasks with 1 item each, due to reduced Serialization and network overhead.
Challenge
Design a performance test suite for a Celery system. Measure: tasks/second (throughput), p50/p95/p99 latency, broker memory usage, worker CPU usage, and result backend storage. Find the optimal concurrency, prefetch, serializer, and batch size for your workload.
FAQ
Mini Project: Performance Benchmark
# benchmark.py
from celery import Celery
import time
import statistics
app = Celery('perf_bench', broker='redis://localhost:6379/0')
@app.task
def perf_task(n):
return n * 2
def run_benchmark(name, config, num_tasks=500):
app.conf.update(config)
print(f"\nBenchmark: {name}")
print(f"Config: {config}")
print("-" * 40)
times = []
for batch in range(3):
start = time.time()
results = [perf_task.delay(i) for i in range(num_tasks)]
for r in results:
r.get(timeout=30)
elapsed = time.time() - start
throughput = num_tasks / elapsed
times.append(elapsed)
print(f" Run {batch + 1}: {elapsed:.2f}s ({throughput:.0f} tasks/s)")
avg_time = statistics.mean(times)
avg_throughput = num_tasks / avg_time
print(f" Average: {avg_time:.2f}s ({avg_throughput:.0f} tasks/s)")
return avg_throughput
# Run benchmarks with different configs
baseline = run_benchmark('Baseline (prefork, concurrency=4)', {
'worker_concurrency': 4,
'worker_pool': 'prefork',
'worker_prefetch_multiplier': 4,
'task_serializer': 'json',
})
tuned = run_benchmark('Tuned (prefork, concurrency=8, prefetch=1)', {
'worker_concurrency': 8,
'worker_pool': 'prefork',
'worker_prefetch_multiplier': 1,
'task_serializer': 'json',
})
fast = run_benchmark('Fast (prefork, concurrency=8, prefetch=1, msgpack)', {
'worker_concurrency': 8,
'worker_pool': 'prefork',
'worker_prefetch_multiplier': 1,
'task_serializer': 'msgpack',
'accept_content': ['msgpack'],
})
print(f"\nImprovement: {fast / baseline:.1f}x over baseline")
Expected output:
Benchmark: Baseline (prefork, concurrency=4)
Run 1: 3.45s (145 tasks/s)
Run 2: 3.51s (142 tasks/s)
Run 3: 3.42s (146 tasks/s)
Average: 3.46s (145 tasks/s)
Benchmark: Tuned (prefork, concurrency=8, prefetch=1)
Run 1: 1.82s (275 tasks/s)
Run 2: 1.78s (281 tasks/s)
Run 3: 1.80s (278 tasks/s)
Average: 1.80s (278 tasks/s)
Improvement: 1.9x over baseline
What's Next
Now that you understand performance tuning, build the mini project: order processing pipeline to apply everything you learned, then explore background jobs for alternative task queue approaches.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro