Celery: Distributed Task Queue

Summary

Celery is a distributed task queue system for Python that enables asynchronous execution of background jobs across multiple workers. It supports scheduling, retries, task workflows, and integrates seamlessly with Django, Flask, and FastAPI.

When to Use

• Background Processing: Offload long-running operations (email, file processing, reports)
• Scheduled Tasks: Cron-like periodic jobs (cleanup, backups, data sync)
• Distributed Computing: Process tasks across multiple workers/servers
• Async Workflows: Chain, group, and orchestrate complex task dependencies
• Real-time Processing: Handle webhooks, notifications, data pipelines
• Load Balancing: Distribute CPU-intensive work across workers

Don't Use When:

• Simple async I/O (use asyncio instead)
• Real-time request/response (use async web frameworks)
• Sub-second latency required (use in-memory queues)
• Minimal infrastructure (use simpler alternatives like RQ or Huey)

Quick Start

Installation

# Basic installation
pip install celery

# With Redis broker
pip install celery[redis]

# With RabbitMQ broker
pip install celery[amqp]

# Full batteries (recommended)
pip install celery[redis,msgpack,auth,cassandra,elasticsearch,s3,sqs]

Basic Setup

# celery_app.py
from celery import Celery

# Create Celery app with Redis broker
app = Celery(
    'myapp',
    broker='redis://localhost:6379/0',
    backend='redis://localhost:6379/1'
)

# Configuration
app.conf.update(
    task_serializer='json',
    accept_content=['json'],
    result_serializer='json',
    timezone='UTC',
    enable_utc=True,
)

# Define a task
@app.task
def add(x, y):
    return x + y

@app.task
def send_email(to, subject, body):
    # Simulate email sending
    import time
    time.sleep(2)
    print(f"Email sent to {to}: {subject}")
    return {"status": "sent", "to": to}

Running Workers

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

# Multiple workers with concurrency
celery -A celery_app worker --concurrency=4 --loglevel=info

# Named worker for specific queues
celery -A celery_app worker -Q emails,reports --loglevel=info

Executing Tasks

# Call task asynchronously
result = add.delay(4, 6)

# Wait for result
print(result.get(timeout=10))  # 10

# Apply async with options
result = send_email.apply_async(
    args=['[email protected]', 'Hello', 'Welcome!'],
    countdown=60  # Execute after 60 seconds
)

# Check task state
print(result.status)  # PENDING, STARTED, SUCCESS, FAILURE

Core Concepts

Architecture Components

Broker: Message queue that stores tasks

• Redis (recommended for most use cases)
• RabbitMQ (enterprise-grade, complex)
• Amazon SQS (serverless, AWS-native)

Workers: Processes that execute tasks

• Pull tasks from broker
• Execute task code
• Store results in backend

Result Backend: Storage for task results

• Redis (fast, in-memory)
• Database (PostgreSQL, MySQL)
• S3 (large results)
• Cassandra, Elasticsearch (specialized)

Beat Scheduler: Periodic task scheduler

• Cron-like scheduling
• Interval-based tasks
• Stores schedule in database or file

Task States

PENDING → STARTED → SUCCESS
                 → RETRY → SUCCESS
                 → FAILURE

• PENDING: Task waiting in queue
• STARTED: Worker picked up task
• SUCCESS: Task completed successfully
• FAILURE: Task raised exception
• RETRY: Task will retry after failure
• REVOKED: Task cancelled before execution

Broker Setup

Redis Configuration

# celery_config.py
broker_url = 'redis://localhost:6379/0'
result_backend = 'redis://localhost:6379/1'

# With authentication
broker_url = 'redis://:password@localhost:6379/0'

# Redis Sentinel (high availability)
broker_url = 'sentinel://localhost:26379;sentinel://localhost:26380'
broker_transport_options = {
    'master_name': 'mymaster',
    'sentinel_kwargs': {'password': 'password'},
}

# Redis connection pool settings
broker_pool_limit = 10
broker_connection_retry = True
broker_connection_retry_on_startup = True
broker_connection_max_retries = 10

RabbitMQ Configuration

# Basic RabbitMQ
broker_url = 'amqp://guest:guest@localhost:5672//'

# With virtual host
broker_url = 'amqp://user:password@localhost:5672/myvhost'

# High availability (multiple brokers)
broker_url = [
    'amqp://user:password@host1:5672//',
    'amqp://user:password@host2:5672//',
]

# RabbitMQ-specific settings
broker_heartbeat = 30
broker_pool_limit = 10

Amazon SQS Configuration

# AWS SQS (serverless)
broker_url = 'sqs://'
broker_transport_options = {
    'region': 'us-east-1',
    'queue_name_prefix': 'myapp-',
    'visibility_timeout': 3600,
    'polling_interval': 1,
}

# With custom credentials
import boto3
broker_transport_options = {
    'region': 'us-east-1',
    'predefined_queues': {
        'default': {
            'url': 'https://sqs.us-east-1.amazonaws.com/123456789/myapp-default',
        }
    }
}

Task Basics

Task Definition

from celery import Task, shared_task
from celery_app import app

# Method 1: Decorator
@app.task
def simple_task(x, y):
    return x + y

# Method 2: Shared task (framework-agnostic)
@shared_task
def framework_task(data):
    return process(data)

# Method 3: Task class (advanced)
class CustomTask(Task):
    def on_success(self, retval, task_id, args, kwargs):
        print(f"Task {task_id} succeeded with {retval}")

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        print(f"Task {task_id} failed: {exc}")

    def on_retry(self, exc, task_id, args, kwargs