Celery, when paired with Redis, allows you to distribute Python tasks across multiple workers, making your applications more scalable and responsive.
Let’s see it in action. Imagine you have a web application that needs to perform a long-running task, like resizing an image or sending a bulk email. Doing this synchronously within the web request would block the user and degrade their experience. Instead, we can offload this to Celery.
First, you need to install Celery and a Redis client:
pip install celery redis
Next, you’ll set up a Celery application. This typically involves a celery_app.py file:
from celery import Celery
# Configure Celery to use Redis as the broker and result backend
app = Celery('tasks',
broker='redis://localhost:6379/0',
backend='redis://localhost:6379/0')
# Optional: Configure task serialization
app.conf.update(
task_serializer='json',
accept_content=['json'],
result_serializer='json',
timezone='UTC',
enable_utc=True,
)
Here, broker is where Celery sends messages (tasks) to be processed, and backend is where results are stored. We’re using Redis on localhost:6379, database 0.
Now, define a task in a separate file, say tasks.py:
from celery_app import app
import time
@app.task
def add(x, y):
"""A simple task that adds two numbers."""
print(f"Adding {x} + {y}...")
time.sleep(5) # Simulate a long-running operation
result = x + y
print(f"Result: {result}")
return result
@app.task
def multiply(x, y):
"""A simple task that multiplies two numbers."""
print(f"Multiplying {x} * {y}...")
time.sleep(3)
result = x * y
print(f"Result: {result}")
return result
To run this, you need a Redis server running. You can start one with Docker:
docker run -d --name my-redis -p 6379:6379 redis
Then, start a Celery worker from your project’s root directory (where celery_app.py and tasks.py are located):
celery -A tasks worker --loglevel=info
The -A tasks points Celery to your task module. --loglevel=info provides helpful output.
From another Python interpreter or your web application, you can now call these tasks asynchronously:
from tasks import add, multiply
# Send tasks to the queue
result_add = add.delay(4, 4)
result_multiply = multiply.delay(5, 3)
print(f"Task add sent. Task ID: {result_add.id}")
print(f"Task multiply sent. Task ID: {result_multiply.id}")
# You can check the status and retrieve results later
# Note: This will block until the result is available
# print(f"Result of add: {result_add.get(timeout=60)}")
# print(f"Result of multiply: {result_multiply.get(timeout=60)}")
The delay() method is a shortcut for apply_async(). It sends the task to Redis, and a Celery worker picks it up and executes the add or multiply function. The output from the worker will show the task being processed.
The magic happens because Redis acts as a message broker. When you call add.delay(), Celery serializes the task details (function name, arguments) and pushes them as a message onto a Redis list. The Celery worker, in a continuous loop, polls Redis for new messages on its assigned queues. When it finds one, it deserializes the message, executes the function with the provided arguments, and then, if a backend is configured, stores the return value or any exceptions in Redis under a key derived from the task ID.
The key levers you control are the broker and backend URLs, which dictate where tasks are queued and results stored. You can also configure task routing, serialization formats, concurrency settings for workers (e.g., -c 4 for 4 concurrent processes), and much more. For instance, you can specify different Redis databases or even use different brokers like RabbitMQ.
When you get a TaskState.RETRY status on a task, it doesn’t necessarily mean the task failed permanently. It implies the task was explicitly told to try again, often due to a transient error like a temporary network issue or a database unavailability that the task logic itself detected and handled by raising a Retry exception with a specified countdown.
The next concept to explore is task routing, which allows you to direct specific tasks to different queues, enabling specialized workers for different types of workloads.