Celery is a distributed task queue framework for Python that enables asynchronous and scheduled task execution using a producer-consumer model. Task are published to a message broker, and worker processes consume and execute them. Celery supports high throughput, fault tolerance through automatic retries, and extensible components.

The core architecture consists of:

• Broker: Message transport (e.g., Redis, RabbitMQ) that queues tasks.
• Worker: Processes that pull tasks from the broker and execute them.
• Backend: Storage for task results (e.g., Redis, database).

Integrating Celery with Django

Django handles requests synchronously, but long-running operations—like sending emails or processing data—should run asynchronously to avoid blocking the main thread. Celery offloads these tasks to background workers.

Setup Steps

1. Install Dependencies

pip install celery redis

2. Configure Celery in Django Project

In myproject/celery.py:

import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')

app = Celery('myproject')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

In myproject/__init__.py:

from .celery import app as celery_app

__all__ = ('celery_app',)

3. Update Django Settings

In settings.py:

CELERY_BROKER_URL = 'redis://127.0.0.1:6379/0'
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/0'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TASK_RESULT_EXPIRES = 86400  # 24 hours
CELERY_TIMEZONE = 'Asia/Shanghai'

4. Define Tasks

In myapp/tasks.py:

from celery import shared_task

@shared_task
def compute_sum(a, b):
    return a + b

5. Trigger Tasks from Views

from django.http import HttpResponse
from .tasks import compute_sum

def trigger_task(request):
    result = compute_sum.delay(7, 9)
    return HttpResponse(f'Task ID: {result.id}')

6. Start Worker

From the project root:

celery -A myproject worker --loglevel=info

7. Retrieve Task Results

from celery.result import AsyncResult
from django.http import JsonResponse

def fetch_result(request):
    task_id = request.GET.get('task_id')
    result = AsyncResult(task_id)
    if result.ready():
        return JsonResponse({
            'status': result.status,
            'output': result.result
        })
    return JsonResponse({'status': result.status, 'output': None})

Scheduled Tasks with Celery Beat

Periodic tasks are configured via CELERY_BEAT_SCHEDULE in settings.py:

from celery.schedules import crontab

CELERY_BEAT_SCHEDULE = {
    'run-every-15-seconds': {
        'task': 'myapp.tasks.compute_sum',
        'schedule': 15.0,
        'args': (3, 5)
    },
    'weekly-report': {
        'task': 'myapp.tasks.generate_report',
        'schedule': crontab(hour=8, minute=0, day_of_week=1),
    }
}

Start the scheduler alongside the worker:

celery -A myproject beat --loglevel=info
celery -A myproject worker --loglevel=info

Task Retry and Context Binding

Tasks can access their execution context when bound:

@shared_task(bind=True, autoretry_for=(Exception,), retry_kwargs={'max_retries': 3, 'countdown': 5})
def fragile_task(self, x, y):
    print(f'Executing {self.request.id}')
    if x < 0:
        raise ValueError("Negative input not allowed")
    return x * y

Alternatively, manual retry logic:

@shared_task(bind=True)
def manual_retry_task(self, value):
    try:
        # Simulate failure
        raise ConnectionError
    except ConnectionError as exc:
        self.retry(exc=exc, countdown=10, max_retries=2)

Task Lifecycle Hooks

Custom behavior on success/failure can be added by subclassing Task:

from celery import Task
import logging

logger = logging.getLogger(__name__)

class LoggedTask(Task):
    def on_success(self, retval, task_id, args, kwargs):
        logger.info(f'Task {task_id} succeeded with result: {retval}')

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        logger.error(f'Task {task_id} failed: {exc}')

@shared_task(base=LoggedTask)
def monitored_task(n):
    return n ** 2

Composing Complex Workflows

Celery supports advanced task patterns:

• Group: Run tasks in parallel.
• Chain: Execute tasks sequentially, passing results forward.
• Chord: Run a group, then trigger a callback with all results.

Example using group:

from celery import group
from .tasks import compute_sum

def batch_operation(request):
    job = group(compute_sum.s(i, i) for i in range(5))
    result = job.apply_async().get()
    return JsonResponse({'results': result})  # [0, 2, 4, 6, 8]

Chain example:

from celery import chain

# Equivalent to mul(mul(2, 3), 4) → 24
workflow = chain(
    compute_mul.s(2, 3),
    compute_mul.s(4)
)
result = workflow().get()

Chord example:

from celery import chord

header = [compute_sum.s(1, 2), compute_mul.s(3, 4)]  # [3, 12]
body = compute_sum.s()  # sum([3, 12]) → 15
result = chord(header)(body).get()

Monitoring with Flower

Flower provides a real-time web dashboard for monitoring workers and tasks.

Install and run:

pip install flower
celery -A myproject flower --port=5555

Access the UI at http://localhost:5555. The HTTP API allows programmatic control:

curl http://localhost:5555/api/tasks