Skip to content
Search

Celery

lastmod 2025-04-08created 2020-04-1316 edits

Celery is an open-source asynchronous task queue, or job queue, based on distributed message passing. It supports scheduling but focuses on real-time operation.

Categories

Installing Celery 

pip install celery

Example

Choosing a Broker

Because Celery is a system that sends and receives tasks (Task) as messages, it needs a separate relay system in the middle acting as a broker (Broker). It is compatible with various Brokers, but I decided to use RabbitMQ, which boasts a long history and stability.

RabbitMQ is a message broker that supports AMQP (Advanced message queuing protocol).

The command below uses Docker to run a RabbitMQ server quickly and easily in a local development environment. 

docker run --hostname my-rabbit \
    -p 5672:5672 \
    -p 8080:15672 \
    -e RABBITMQ_DEFAULT_USER=guest \
    -e RABBITMQ_DEFAULT_PASS=guest \
    --name some-rabbit \
    rabbitmq:3-management

Creating a Task

The first thing to do is define a Celery instance. This is usually called the Celery App, and it is the entry point for everything Celery does, such as creating Tasks and managing Workers.

Pass the module name as the first positional argument of the Celery Class, and pass the Broker connection information as the broker keyword argument. 

# tasks.py
from celery import Celery

app = Celery('tasks', broker='pyamqp://guest:guest@localhost//')

Next, write a simple Task function. Since this is a simple practice step, the Celery instance and the Task function are defined together in the same Python module (tasks.py). 

# tasks.py

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

Running the Celery Woker

Installing Celery automatically makes the celery command available. So to start a Celery Worker, build the base command in the form $ celery worker, and specify the Python module containing the Celery instance defined above with the -A {module name} or --app={module name} parameter. 

celery -A tasks worker --loglevel=info

Calling a Task

A Celery Task is ultimately a Python function too. A Python function is normally called like add(), but to call it as a Celery Task, call the delay() method in the form add.delay(). (This method is a shortcut version of the apply_async() method.) 

from tasks import add
add.delay(1, 2)

Calling a Task returns an #AsyncResult instance. This object can be used to check the state, such as whether the Task has completed or returned a result.

By default, Task results are not stored. To store results in a DB or to make RPC calls, you need to add a result backend setting.

Keeping task results (Result Backend)

Celery has several built-in result backends for storing Task results. (e.g., SQLAlchemy, Django, Memcached, Redis, RPC, etc.)

To specify a result backend, add the backend keyword argument to the Celery instance.

When rpc is specified as the result backend, it works by sending task results back as transient AMQP messages, and the AsyncResult instance returned as the task result can be kept. 

app = Celery('tasks', backend='rpc://', broker='pyamqp://')

Use the ready() method to check whether a Task has completed. 

result = add.delay(4, 4)

result.ready()
# True

The get() method retrieves the result of a completed task. If the task has not completed at the time of the call, it waits until it completes. (Reportedly this is rarely used.) 

result.get()

If an exception occurs while a Task is running, the exception can be propagated through the get() method. Passing False as the propagate argument prevents the exception from being propagated. 

result.get(propagate=False)

Applying the task(ignore_result=True) decorator to a Task function disables the result backend for that individual Task. 

@app.task(ignore_result=True)
def subtraction(x, y):
    return x - y

Managing Celery configuration separately

Running Celery does not require many settings, but a Broker connection is mandatory and a result backend is optional. In the example above, the basic settings were passed directly to the Celery instance, but in larger projects it is easier to control settings by defining them in a separate dedicated module (py). More settings can be found in Configuration and defaults.

Once the settings are separated into a dedicated module, you can pass the configuration module to the Celery instance with the app.config_from_object() method. 

app.config_from_object('celeryconfig')

celeryconfig.py file: 

broker_url = 'pyamqp://'
result_backend = 'rpc://'

task_serializer = 'json'
result_serializer = 'json'
accept_content = ['json']
timezone = 'Asia/Seoul'
enable_utc = True

# 오작동 한 작업을 전용 대기열로 라우팅하는 설정
task_routes = {
    'tasks.add': 'low-priority'
}

# 작업 속도를 제한하는 설정
task_annotations = {
    'tasks.add': {'rate_limit': '10/m'
}

How to select a specific worker

In short, split the Message Queue.

Using Amazon SQS

AsyncResult

See also

Favorite site

author Zer0 sitemap