Celery Architecture Overview

Celery employs a three-componant architecture: message broker, task execution workers, and result storage backend.

Basic Implementation

Create a task module named task_manager.py:

import celery
import time

result_backend = 'redis://localhost:6379/1'
message_broker = 'redis://localhost:6379/2'

app = celery.Celery('demo_app', backend=result_backend, broker=message_broker)

@app.task
def process_notification(recipient):
    print(f"Sending notification to {recipient}...")
    time.sleep(3)
    print(f"Notification to {recipient} completed")
    return "success"

Execute the worker processs with:

celery -A task_manager worker --loglevel=info --pool=eventlet

Task Invocation

Create task_dispatcher.py to trigger tasks:

from task_manager import process_notification

task1 = process_notification.delay("user_a")
print(f"Task ID: {task1.id}")

task2 = process_notification.delay("user_b")
print(f"Task ID: {task2.id}")

Result Retrieval

Check task outcomes with result_checker.py:

from celery.result import AsyncResult
from task_manager import app

task_result = AsyncResult(id="your-task-id-here", app=app)

if task_result.successful():
    outcome = task_result.get()
    print(f"Result: {outcome}")
elif task_result.failed():
    print("Task execution failed")
elif task_result.state == 'PENDING':
    print("Task awaiting execution")
elif task_result.state == 'RETRY':
    print("Task retrying after failure")
elif task_result.state == 'STARTED':
    print("Task currently executing")

Multi-Module Configuration

Configure multiple task modules in celery_config.py:

from celery import Celery

app = Celery('multi_task_app',
             broker='redis://localhost:6379/1',
             backend='redis://localhost:6379/2',
             include=['tasks.module_a', 'tasks.module_b'])

app.conf.timezone = 'Asia/Shanghai'
app.conf.enable_utc = False

Scheduled Tasks

Implement timed task execution:

from task_manager import process_notification
from datetime import datetime, timedelta

current_time = datetime.now()
utc_time = datetime.utcfromtimestamp(current_time.timestamp())
delay = timedelta(seconds=15)
execution_time = utc_time + delay

scheduled_task = process_notification.apply_async(
    args=["scheduled_user"], eta=execution_time
)
print(f"Scheduled task ID: {scheduled_task.id}")

Periodic Task Configuration

Configure recurring tasks in celery_config.py:

from datetime import timedelta
from celery import Celery

app = Celery('periodic_tasks', 
             broker='redis://localhost:6379/1',
             backend='redis://localhost:6379/2',
             include=['tasks.notifications', 'tasks.reports'])

app.conf.timezone = 'Asia/Shanghai'
app.conf.enable_utc = False

app.conf.beat_schedule = {
    'regular-notification': {
        'task': 'tasks.notifications.process_notification',
        'schedule': timedelta(seconds=10),
        'args': ('recurring_user',)
    }
}

Service Initialization

Start the scheduler service:

celery beat -A celery_config

Launch worker processes:

celery -A celery_config worker -l info

Alternative, combine both functions:

celery -B -A celery_config worker -l info