Celery Worker

Celery Worker Skill

Production-ready Celery worker configuration for distributed task processing.

When to Use This Skill

Use this skill when asked to:

• Configure Celery workers for background tasks
• Set up task queues and routing
• Configure retry strategies
• Optimize worker performance
• Debug Celery issues

Architecture Overview

┌─────────────────────────────────────────────────────────────┐
│                    FastAPI Application                       │
│  dispatch_to_celery() → task.delay() → Redis Broker         │
└─────────────────────────────┬───────────────────────────────┘
                              │
                              ▼
                    ┌─────────────────┐
                    │   Redis Broker  │
                    │   (Message Queue)│
                    └────────┬────────┘
                             │
              ┌──────────────┼──────────────┐
              │              │              │
              ▼              ▼              ▼
        ┌──────────┐   ┌──────────┐   ┌──────────┐
        │ Worker 1 │   │ Worker 2 │   │ Worker 3 │
        │ Q: celery│   │ Q: files │   │ Q: email │
        └──────────┘   └──────────┘   └──────────┘
                              │
                              ▼
                    ┌─────────────────┐
                    │ Result Backend  │
                    │    (Redis)      │
                    └─────────────────┘

Celery App Configuration

# celery_app.py
from celery import Celery
from settings import settings

celery_app = Celery(
    "app_name",
    broker=settings.CELERY_BROKER_URL,
    backend=settings.CELERY_RESULT_BACKEND,
    include=[
        "tasks.email",
        "tasks.files",
        "tasks.scheduler",
    ],
)

celery_app.conf.update(
    # Serialization
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    
    # Reliability - CRITICAL
    task_acks_late=True,              # Ack after completion
    task_reject_on_worker_lost=True,  # Requeue if worker dies
    task_track_started=True,          # Track task start
    
    # Results
    result_expires=86400,             # 24 hours
    
    # Worker Performance
    worker_prefetch_multiplier=1,     # Fair distribution
    worker_concurrency=10,            # Concurrent tasks
    
    # Time Limits - IMPORTANT
    task_soft_time_limit=300,         # 5 min soft limit
    task_time_limit=360,              # 6 min hard limit
    
    # Retries
    task_default_retry_delay=60,      # 1 min default delay
    
    # Timezone
    timezone="UTC",
    enable_utc=True,
    
    # Connection
    broker_connection_retry_on_startup=True,
)

Task Definition Pattern

@shared_task(
    bind=True,                    # Access self
    max_retries=3,                # Retry 3 times
    default_retry_delay=60,       # 1 min delay
    autoretry_for=(Exception,),   # Auto-retry
    retry_backoff=True,           # Exponential backoff
    retry_backoff_max=300,        # Max 5 min
    retry_jitter=True,            # Prevent thundering herd
    soft_time_limit=120,          # Soft limit
    time_limit=180,               # Hard limit
)
def my_task(self, **kwargs):
    pass

Worker Commands

# Basic worker
celery -A celery_app worker --loglevel=info

# With concurrency
celery -A celery_app worker --loglevel=info --concurrency=4

# Specific queues
celery -A celery_app worker -Q celery,file_queue,email_queue

# With gevent pool (high concurrency)
celery -A celery_app worker -P gevent --concurrency=100

Queue Routing

# celery_app.py
celery_app.conf.task_routes = {
    'tasks.email.*': {'queue': 'email'},
    'tasks.files.*': {'queue': 'files'},
    'tasks.heavy.*': {'queue': 'heavy'},
}

# Or per-task
@shared_task(queue='high_priority')
def urgent_task():
    pass

Key Patterns

1. task_acks_late=True - Acknowledge after completion for reliability
2. Time limits - Prevent hung tasks
3. Retry with backoff - Handle transient failures
4. Queue routing - Separate workloads
5. Result expiration - Clean up old results