Running the Celery Worker — Complete Guide
In this tutorial, you will learn about Running the Celery Worker. We cover key concepts, practical examples, and best practices to help you master this topic.
Run Celery workers with different concurrency models, configure worker settings, manage worker pools, and handle graceful shutdown for production deployments.
What You Learn
You will learn how to start Celery workers with various flags, configure concurrency (prefork, gevent, solo), manage worker pools, handle shutdown, and run workers as daemons.
Why It Matters
Worker configuration directly impacts throughput, latency, and reliability. An undersized worker pool causes task backlog. An oversized pool exhausts memory. Improper shutdown causes lost tasks. Correct worker management is essential for production.
Real-World Use
DodaTech runs Celery workers with gevent concurrency for I/O-bound tasks (HTTP requests, file I/O) and prefork for CPU-bound tasks (file scanning). Workers are configured with --max-tasks-per-child=1000 to prevent memory leaks.
Starting a Worker
# Basic worker
celery -A tasks worker --loglevel=info
# Worker with name
celery -A tasks worker --loglevel=info --hostname=worker1@%h
# Worker with specific concurrency
celery -A tasks worker --concurrency=8 --loglevel=info
# tasks.py
from celery import Celery
app = Celery('demo', broker='redis://localhost:6379/0')
@app.task
def echo(message):
return f"Worker says: {message}"
Concurrency Models
flowchart TB
subgraph "Concurrency Models"
PF[prefork: Process pool
CPU-bound tasks]
GV[gevent: Green threads
I/O-bound tasks]
SL[solo: Single process
Debugging only]
end
PF --> |pros| CP[True parallelism]
PF --> |cons| CM[High memory]
GV --> |pros| GM[Low memory]
GV --> |cons| GC[No CPU parallelism]
SL --> |pros| SD[Simple debugging]
SL --> |cons| SC[No concurrency]
# Prefork (default) - best for CPU-bound tasks
celery -A tasks worker --concurrency=4 --pool=prefork
# Gevent - best for I/O-bound tasks
celery -A tasks worker --concurrency=100 --pool=gevent
# Solo - single process, for debugging
celery -A tasks worker --pool=solo --loglevel=debug
Worker Options
# Common worker options
celery -A tasks worker \
--loglevel=info \ # Log level
--concurrency=8 \ # Number of worker processes/threads
--pool=prefork \ # Pool implementation
--hostname=worker1@%h \ # Worker name (%h = hostname)
--queues=default,high \ # Queues to consume from
--max-tasks-per-child=1000 \ # Restart after N tasks
--max-memory-per-child=50000 \ # Restart if >50MB memory
--time-limit=300 \ # Hard time limit
--soft-time-limit=240 \ # Soft time limit
--autoscale=10,3 \ # Max=10, Min=3 workers
--beat \ # Run Celery Beat in same process
--without-gossip \ # Disable gossip (reduce chatter)
--without-mingle \ # Skip mingle phase
--without-heartbeat \ # Disable heartbeats
--events \ # Send task events
--task-events \ # Send task-specific events
--discard \ # Discard all pending tasks
--purge \ # Purge all configured queues
Autoscaling
Workers can dynamically scale based on load:
# Autoscale between 2 and 10 workers
celery -A tasks worker --autoscale=10,2 --loglevel=info
from celery import Celery
app = Celery('autoscale', broker='redis://localhost:6379/0')
app.conf.update(
worker_autoscaler='celery.worker.autoscale:Autoscaler',
worker_max_tasks_per_child=1000,
)
@app.task
def scale_demo(data):
return f"Processed: {data}"
Running as a Daemon
Using systemd:
# /etc/systemd/system/celery.service
[Unit]
Description=Celery Worker Service
After=network.target redis.service
[Service]
Type=forking
User=celery
Group=celery
EnvironmentFile=/etc/celery/celery.conf
WorkingDirectory=/opt/myapp
ExecStart=/bin/sh -c '${CELERY_BIN} multi start ${CELERYD_NODES} \
-A ${CELERY_APP} --pidfile=${CELERYD_PID_FILE} \
--logfile=${CELERYD_LOG_FILE} --loglevel=${CELERYD_LOG_LEVEL} \
${CELERYD_OPTS}'
ExecStop=/bin/sh -c '${CELERY_BIN} multi stopwait ${CELERYD_NODES} \
--pidfile=${CELERYD_PID_FILE}'
ExecReload=/bin/sh -c '${CELERY_BIN} multi restart ${CELERYD_NODES} \
-A ${CELERY_APP} --pidfile=${CELERYD_PID_FILE} \
--logfile=${CELERYD_LOG_FILE} --loglevel=${CELERYD_LOG_LEVEL} \
${CELERYD_OPTS}'
Restart=on-failure
[Install]
WantedBy=multi-user.target
# /etc/celery/celery.conf
CELERY_BIN=/usr/local/bin/celery
CELERY_APP=tasks
CELERYD_NODES="worker1 worker2 worker3"
CELERYD_OPTS="--concurrency=8 --queues=default,high --events"
CELERYD_LOG_LEVEL=INFO
CELERYD_LOG_FILE=/var/log/celery/%n.log
CELERYD_PID_FILE=/var/run/celery/%n.pid
Graceful Shutdown
# Warm shutdown (finishes current tasks)
celery multi stop worker1 --pidfile=/var/run/celery/%n.pid
# Cold shutdown (immediate)
celery multi stopwait worker1 --pidfile=/var/run/celery/%n.pid
# Restart
celery multi restart worker1 --pidfile=/var/run/celery/%n.pid
Worker Status
# Inspect active workers
celery -A tasks inspect active
celery -A tasks inspect scheduled
celery -A tasks inspect reserved
celery -A tasks inspect registered
celery -A tasks inspect stats
celery -A tasks status
Expected output:
celery@worker1: OK
celery@worker2: OK
celery@worker3: OK
from tasks import app
# Check worker status from Python
inspect = app.control.inspect()
print(f"Active workers: {inspect.ping()}")
print(f"Active tasks: {inspect.active()}")
print(f"Registered tasks: {inspect.registered()}")
Common Mistakes
1. Running Workers Without Enough Concurrency
A single worker process runs one task at a time. Set concurrency to at least 4-8 for production. Monitor CPU usage to find the sweet spot.
2. Not Setting max-tasks-per-child
Memory leaks in tasks accumulate over time. Set --max-tasks-per-child=1000 to restart worker processes periodically.
3. Using solo Pool in Production
The solo pool runs one task at a time with no concurrency. It is for debugging only. Use prefork or gevent in production.
4. Ignoring Worker Logs
Worker logs show task failures, retries, and timing. Monitor them. Set up log aggregation with tools like the ELK Stack.
5. Running Workers Without Events
Without --events, monitoring tools like Flower cannot see task activity. Always enable events in production.
Practice Questions
1. What is the default concurrency model in Celery?
prefork (process pool). It creates N worker processes that each execute tasks independently, providing true parallelism for CPU-bound tasks.
2. When should you use gevent instead of prefork?
When tasks are I/O-bound (HTTP requests, database queries, file reads). Gevent uses green threads that yield on I/O, allowing many concurrent tasks with low memory overhead.
3. What does --max-tasks-per-child do?
It restarts a worker process after it has executed N tasks. This prevents memory leaks from accumulating over time.
4. How do you inspect active worker status?
Use celery -A app inspect active or the Python API app.control.inspect(). These show running, scheduled, and reserved tasks.
Challenge
Design a worker deployment Strategy for a platform with 3 task types: CPU-intensive (video encoding), I/O-intensive (API calls), and mixed (database operations). Choose appropriate concurrency models, pool sizes, and worker separation for each type.
FAQ
Mini Project: Worker Fleet Manager
# fleet_manager.py
import subprocess
import json
import time
import signal
import os
class CeleryFleet:
def __init__(self, app_name='tasks', num_workers=3):
self.app_name = app_name
self.num_workers = num_workers
self.processes = {}
def start_workers(self):
for i in range(self.num_workers):
worker_name = f"worker_{i}"
cmd = [
'celery', '-A', self.app_name, 'worker',
f'--hostname={worker_name}@%h',
'--concurrency=4',
'--loglevel=info',
'--events',
'--max-tasks-per-child=1000',
]
process = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True
)
self.processes[worker_name] = process
print(f"Started {worker_name} (PID: {process.pid})")
def stop_workers(self):
for name, process in self.processes.items():
process.terminate()
print(f"Stopping {name} (PID: {process.pid})")
for name, process in self.processes.items():
process.wait(timeout=10)
print(f"Stopped {name}")
def get_status(self):
from tasks import app
inspect = app.control.inspect()
try:
workers = inspect.ping()
if workers:
for worker, status in workers.items():
print(f"{worker}: {status}")
else:
print("No workers responding")
except Exception as e:
print(f"Error: {e}")
if __name__ == '__main__':
fleet = CeleryFleet()
try:
fleet.start_workers()
time.sleep(5)
fleet.get_status()
time.sleep(30)
finally:
fleet.stop_workers()
Expected output:
Started worker_0 (PID: 12345)
Started worker_1 (PID: 12346)
Started worker_2 (PID: 12347)
celery@worker_0: {'ok': 'pong'}
celery@worker_1: {'ok': 'pong'}
celery@worker_2: {'ok': 'pong'}
Stopping worker_0 (PID: 12345)
Stopping worker_1 (PID: 12346)
Stopping worker_2 (PID: 12347)
Stopped worker_0
Stopped worker_1
Stopped worker_2
What's Next
Now that you know how to run workers, learn about calling tasks with delay and apply_async, then explore task retry and error handling for building robust task pipelines.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro