Celery is a powerful framework for distributed asynchronous task queues, allowing tasks to execute independently of the main program and even run on other hosts. It is commonly used to implement asynchronous tasks and scheduled tasks.

Key concepts in Celery include Broker and Backend.

Broker

A broker acts as a message transfer intermediary or message queue, functioning like an email box. When an application calls an asynchronous task in Celery, it sends a message to the broker, which is then picked up by a Celery worker for execution. The term 'broker' translates to 'agent' and refers to a message queue used for sending and receiving messages. Options include RabbitMQ (message queue), Redis (cache database), and databases (not recommended).

Backend

The backend stores messages and results related to Celery's execution. It is configured using CELERY_RESULT_BACKEND, which saves results and status. If tracking task status is needed, this should be set. It can be a database or cache backend, among others.

Brokers are typically RabbitMQ and Redis, while backends often use databases. For simplicity, Redis can be used for both.

The architecture of Celery consists of three components: message broker, worker, and task result store.

1. Message Broker

Celery does not provide message services itself but integrates easily with third-party message middleware. This includes RabbitMQ, Redis, MongoDB (experimental), Amazon SQS (experimental), CouchDB (experimental), SQLAlchemy (experimental), Django ORM (experimental), IronMQ.

2. Worker

Workers are the units responsible for executing tasks, running concurrently across distributed system nodes.

3. Task Result Store

This stores the results of tasks executed by workers. Celery supports storing results in various ways, including AMQP, Redis, Memcached, MongoDB, SQLAlchemy, Django ORM, Apache Cassandra, IronCache, etc.

Installation:

pip install redis
pip install celery

The architecture is shown below.

Celery comprises several modules:

• Task Module

  Contains asynchronous and scheduled tasks. Asynchronous tasks are triggered within business logic and sent to the task queue, while scheduled tasks are periodically sent to the task queue by the Celery Beat processs.

• Message Broker

  Acts as a task scheduling queue, receiving messages (tasks) from producers and storing them in the queue. Celery does not provide queue service, and the official recommendation is to use RabbitMQ and Redis.

• Worker

  Executes tasks, monitoring the message queue for tasks to process.

• Backend

  Stores task execution results for querying. Like the message broker, storage can also use RabbitMQ, Redis, and MongoDB.

Using Celery

Using Celery involves three aspects: defining task functions, running the Celery service, and calling from the client application.

Creating a Celery Instance

Save the following code in a file named tasks.py:

# -*- coding: utf-8 -*-

import time
from celery import Celery

broker = 'redis://127.0.0.1:6379'
backend = 'redis://127.0.0.1:6379/0'

app = Celery('my_task', broker=broker, backend=backend)

@app.task
def add(x, y):
    time.sleep(5)     # Simulate a time-consuming operation
    return x + y

The code above performs the following:

Creates a Celery instance named my_task;

Specifies Redis as the message broker with URL redis://127.0.0.1:6379;

Specifies Redis as the backend with URL redis://127.0.0.1:6379/0;

Creates a Celery task add, which becomes a schedulable task after being decorated with @app.task;

Starting the Celery Worker

In the current directory, start the Celery Worker using the following command:

celery worker -A tasks --loglevel=info

Parameters:

-A specifies the location of the Celery instance, which is in tasks.py. Celery automatically finds the Celery object instance in the file. You can also specify it manually, such as -A tasks.app; --loglevel sets the log level, defaulting to warning, and can use -l info;

In production environments, Supervisor is usually used to manage the Celery Worker process.

Upon successful startup, the console displays output similar to:

Calling the Task

You can call the task using the delay() or apply_async() methods in the application.

Open the Python console in the current directory and enter the following code:

>>> from tasks import add
>>> add.delay(2, 8)
<AsyncResult: 2272ddce-8be5-493f-b5ff-35a0d9fe600f>

Here, we imported the add task object from the tasks.py file and used the delay() method to send the task to the message broker (Broker). After the Celery Worker process detects the task, it will execute it. Switching to the Worker startup window shows additional logs:

[2016-12-10 12:00:50,376: INFO/MainProcess] Received task: tasks.add[2272ddce-8be5-493f-b5ff-35a0d9fe600f]
[2016-12-10 12:00:55,385: INFO/PoolWorker-4] Task tasks.add[2272ddce-8be5-493f-b5ff-35a0d9fe600f] succeeded in 5.00642602402s: 10

This indicates that the task has been scheduled and executed successfully.

To get the result after execution, do the following:

>>> result = add.delay(2, 6)
>>> result.ready()   # Use ready() to check if the task is completed
False
>>> result.ready()
False
>>> result.ready()
True
>>> result.get()     # Use get() to retrieve the task result
8

In the above example, we called the task within the Python environment. Typically, you would call the task within an application. For example, save the following code in client.py:

# -*- coding: utf-8 -*-

from tasks import add

# Asynchronous task
add.delay(2, 8)

print 'hello world'

Running the command python client.py shows that although the add function needs to wait 5 seconds before returning the result, since it is an asynchronous task, it does not block the main program. Therefore, the main program continues to execute the print statement and outputs the result.

Using Configuration

In the previous example, the Broker and Backend configurations were directly written into the program. A better approach is to place all configuration items in a single configuration file, usually named celeryconfig.py. Celery has many configurations, and their meanings can be found in the official documentation.

Below is another example. The project structure is as follows:

celery_demo                    # Project root directory
    ├── celery_app             # Store celery-related files
    │   ├── __init__.py
    │   ├── celeryconfig.py    # Configuration file
    │   ├── task1.py           # Task file 1
    │   └── task2.py           # Task file 2
    └── client.py              # Application

The __init__.py code is as follows:

# -*- coding: utf-8 -*-

from celery import Celery

app = Celery('demo')                                # Create a Celery instance
app.config_from_object('celery_app.celeryconfig')   # Load the configuration module through the Celery instance

The celeryconfig.py code is as follows:

BROKER_URL = 'redis://127.0.0.1:6379'               # Specify the Broker
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/0'  # Specify the Backend

CELERY_TIMEZONE='Asia/Shanghai'                     # Specify the timezone, default is UTC
# CELERY_TIMEZONE='UTC'                              

CELERY_IMPORTS = (                                  # Specify the imported task modules
    'celery_app.task1',
    'celery_app.task2'
)

The task1.py code is as follows:

import time
from celery_app import app

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

The task2.py code is as follows:

import time
from celery_app import app

@app.task
def multiply(x, y):
    time.sleep(2)
    return x * y

The client.py code is as follows:

# -*- coding: utf-8 -*-

from celery_app import task1
from celery_app import task2

task1.add.apply_async(args=[2, 8])        # Also can use task1.add.delay(2, 8)
task2.multiply.apply_async(args=[3, 7])   # Also can use task2.multiply.delay(3, 7)

print 'hello world'

Now, start the Celery Worker process in the project root directory by running the following command:

celery -A celery_app worker --loglevel=info

Then run python client.py. It will send two asynchronous tasks to the Broker. In the Worker window, you can see the following output:

[2016-12-10 13:51:58,939: INFO/MainProcess] Received task: celery_app.task1.add[9ccffad0-aca4-4875-84ce-0ccfce5a83aa]
[2016-12-10 13:51:58,941: INFO/MainProcess] Received task: celery_app.task2.multiply[64b1f889-c892-4333-bd1d-ac667e677a8a]
[2016-12-10 13:52:00,948: INFO/PoolWorker-3] Task celery_app.task1.add[9ccffad0-aca4-4875-84ce-0ccfce5a83aa] succeeded in 2.00600231002s: 10
[2016-12-10 13:52:00,949: INFO/PoolWorker-4] Task celery_app.task2.multiply[64b1f889-c892-4333-bd1d-ac667e677a8a] succeeded in 2.00601326401s: 21

Scheduled Tasks

Besides executing asynchronous tasks, Celery also supports periodic tasks (scheduled tasks). The Celery Beat process reads the configuration file content and periodically sends scheduled tasks to the task queue.

Let's look at an example. The project structure is as follows:

celery_demo                    # Project root directory
    ├── celery_app             # Store celery-related files
        ├── __init__.py
        ├── celeryconfig.py    # Configuration file
        ├── task1.py           # Task file
        └── task2.py           # Task file

The __init__.py code is as follows:

# -*- coding: utf-8 -*-

from celery import Celery

app = Celery('demo')
app.config_from_object('celery_app.celeryconfig')

The celeryconfig.py code is as follows:

# -*- coding: utf-8 -*-

from datetime import timedelta
from celery.schedules import crontab

# Broker and Backend
BROKER_URL = 'redis://127.0.0.1:6379'
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/0'

# Timezone
CELERY_TIMEZONE='Asia/Shanghai'    # Specify the timezone, default is 'UTC'
# CELERY_TIMEZONE='UTC'

# imports
CELERY_IMPORTS = (
    'celery_app.task1',
    'celery_app.task2'
)

# schedules
CELERYBEAT_SCHEDULE = {
    'add-every-30-seconds': {
         'task': 'celery_app.task1.add',
         'schedule': timedelta(seconds=30),       # Execute every 30 seconds
         'args': (5, 8)                           # Task function parameters
    },
    'multiply-at-some-time': {
        'task': 'celery_app.task2.multiply',
        'schedule': crontab(hour=9, minute=50),   # Execute daily at 9:50 AM
        'args': (3, 7)                            # Task function parameters
    }
}

The task1.py code is as follows:

import time
from celery_app import app

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

The task2.py code is as follows:

import time
from celery_app import app

@app.task
def multiply(x, y):
    time.sleep(2)
    return x * y

Now, start the Celery Worker process in the project root directory by running the following command:

celery -A celery_app worker --loglevel=info

Next, start the Celery Beat process to schedule tasks to the Broker, in the project root directory run the following command:

celery_demo $ celery beat -A celery_app
celery beat v4.0.1 (latentcall) is starting.
__    -    ... __   -        _
LocalTime -> 2016-12-11 09:48:16
Configuration ->
    . broker -> redis://127.0.0.1:6379//
    . loader -> celery.loaders.app.AppLoader
    . scheduler -> celery.beat.PersistentScheduler
    . db -> celerybeat-schedule
    . logfile -> [stderr]@%WARNING
    . maxinterval -> 5.00 minutes (300s)

After that, in the Worker window, you can see that task1 runs every 30 seconds, and task2 runs once daily at 9:50 AM.

In the above example, two commands were used to start the Worker and Beat processes. They can also be combined into one command:

$ celery -B -A celery_app worker --loglevel=info

Error: Celery ValueError: not enough values to unpack (expected 3, got 0)

Solution:

pip install eventlet

Run: celery -A worker -l info -P eventlet

If -P eventlet is not added, the eror will still occur.