Comprehensive Guide to Using Celery for Asynchronous Task Processing in Python

Celery is introduced with official links ( http://www.celeryproject.org/, http://docs.celeryproject.org/en/latest/index.html) and a brief overview of its role as an asynchronous task framework.

The architecture consists of three parts: a message broker (e.g., RabbitMQ or Redis), worker processes that execute tasks, and a result store for task outcomes.

Typical usage scenarios include asynchronous execution for time‑consuming jobs, delayed execution with apply_async, and periodic execution via Celery beat.

Installation is performed with pip install celery, followed by configuring the broker and backend, for example:

broker = 'redis://127.0.0.1:6379/1'
backend = 'redis://127.0.0.1:6379/2'
app = Celery(broker=broker, backend=backend, include=['celery_task.tasks'])

Two recommended project structures are shown. When using a package, create a celery_task directory containing __init__.py, celery.py (the Celery app), and tasks.py (task definitions). When using a single module, place the Celery app directly in celery.py.

Basic task examples illustrate how to define and register tasks:

@app.task
def add(n1, n2):
    return n1 + n2

@app.task
def low(n1, n2):
    return n1 - n2

Adding tasks can be done immediately with tasks.add.delay(10, 20) or delayed with apply_async and an eta value, e.g.:

t1 = tasks.add.delay(10, 20)
eta_time = datetime.utcnow() + timedelta(seconds=10)
result = tasks.some_task.apply_async(args=(a, b), eta=eta_time)

Results are retrieved using AsyncResult:

from celery.result import AsyncResult
async_res = AsyncResult(id='task-id', app=app)
if async_res.successful():
    print(async_res.get())

Advanced usage demonstrates configuring periodic tasks with Celery beat, setting timezone, disabling UTC, and defining a schedule dictionary:

app.conf.timezone = 'Asia/Shanghai'
app.conf.enable_utc = False
app.conf.beat_schedule = {
    'update_banner': {
        'task': 'celery_task.tasks.update_banner_list',
        'schedule': timedelta(seconds=10),
    },
}

Integration with Django includes Redis cache settings, a viewset that caches banner data, and a Celery task that updates the cache:

cache.set('banner_list', banner_list, 86400)

@app.task
def update_banner_list():
    queryset = Banner.objects.filter(is_delete=False, is_show=True).order_by('-orders')[:settings.BANNER_COUNT]
    banner_list = BannerSerializer(queryset, many=True).data
    for banner in banner_list:
        banner['image'] = f'http://127.0.0.1:8000{banner["image"]}'
    cache.set('banner_list', banner_list, 86400)
    return True

Finally, the article explains how to start the services: run a worker with celery worker -A celery_task -l info -P eventlet, start the beat scheduler with celery beat -A celery_task -l info, and launch the Django server to serve cached endpoints.