KAHIBARO
Discord Login Register ☰
☰

17.5. Redis as a Message Broker

Table of Contents

Why Use Redis as a Message Broker?

In background processing, a message broker is the part that sits between the web application and the workers. It receives jobs, stores them, and delivers them to workers.

Redis is a very popular choice for this role because:

You already saw Redis used for caching and other use cases. Here we focus on using Redis specifically as the transport layer for background jobs.

Important rule: A message broker should be reliable. Your application must never assume that "fire and forget" is enough. Always design for:

  • Jobs being retried.
  • Messages possibly being delivered more than once.
  • Workers possibly crashing while processing.

Basic Message Queue Concepts with Redis

A queue is a list of tasks that need to be processed. The main operations are:

With Redis you can implement queues using list commands like LPUSH, RPUSH, LPOP, and blocking operations like BLPOP.

Simple Redis Queue with Lists

Imagine a queue called email_queue that stores tasks. Each task is a JSON string.

Adding a message (producer)

Producer (web app) enqueues a new task:

bash 
LPUSH email_queue '{"to": "alice@example.com", "subject": "Hello", "body": "Hi Alice"}'

Or from Python:

python 
import json
import redis
r = redis.Redis(host="localhost", port=6379, db=0)
task = {
    "to": "alice@example.com",
    "subject": "Hello",
    "body": "Hi Alice",
}
r.lpush("email_queue", json.dumps(task))

Here the left side of the list is considered the "front" of the queue.

Consuming a message (worker)

A worker can block and wait for new jobs:

bash 
BLPOP email_queue 0

The 0 means "wait forever until something is available".

In Python:

python 
import json
import redis
r = redis.Redis(host="localhost", port=6379, db=0)
while True:
    # BLPOP returns (queue_name, data)
    _, raw_task = r.blpop("email_queue", timeout=0)
    task = json.loads(raw_task)
    print(f"Sending email to {task['to']}")
    # ... actually send email here ...

This is already a working message queue with:

But it has a major limitation: if the worker crashes after BLPOP but before finishing the job, the message is lost.

Reliable Message Delivery Patterns

To be a useful message broker, Redis needs to support reliable processing:

Redis on its own does not give you a full job system, but you can build patterns on top of it or use libraries that do it for you.

Using a Work Queue and a Processing Queue

A common pattern is:

The idea:

  1. Worker moves a job from pending to processing in one atomic step using RPOPLPUSH.
  2. Worker processes the job.
  3. If processing succeeds, worker removes the job from processing.
  4. If worker crashes, the job remains in processing. A recovery process can check for old jobs in processing and requeue them.

Step-by-step with Redis commands

1. Producer puts job in pending queue
bash 
LPUSH queue:pending '{"id": 1, "task": "send_email", "to": "alice@example.com"}'
2. Worker atomically moves job to processing
bash 
RPOPLPUSH queue:pending queue:processing

This removes the rightmost element from queue:pending and pushes it to the left of queue:processing, atomically.

In Python:

python 
raw_job = r.rpoplpush("queue:pending", "queue:processing")
job = json.loads(raw_job)
3. Worker processes the job
python 
try:
    # Do the work
    print(f"Sending email to {job['to']}")
    # ... send email ...
except Exception:
    # If failed, you may choose to move it back or log
    raise
else:
    # If successful, remove from processing
    r.lrem("queue:processing", 1, raw_job)
4. Recovery process

A separate script can scan queue:processing and decide if items are too old and should be requeued:

python 
processing_jobs = r.lrange("queue:processing", 0, -1)
for raw_job in processing_jobs:
    job = json.loads(raw_job)
    # Suppose job has a "started_at" timestamp
    # If too old, move back to pending

This pattern is the foundation of many Redis-based queues.

Rule: Never just BLPOP from a single queue if you care about reliability. Use patterns that keep a separate processing state or use a library that implements it for you.

Using Redis Pub/Sub vs Queues

Redis also has Pub/Sub, which allows clients to subscribe to channels and receive messages that are published to them.

Example:

bash 
SUBSCRIBE log_channel
PUBLISH log_channel "Server restarted"

In Python:

python 
import redis
r = redis.Redis()
pubsub = r.pubsub()
pubsub.subscribe("log_channel")
for message in pubsub.listen():
    print(message)

Differences from queues:

FeatureQueues (lists)Pub/Sub
Message persistenceStored in Redis until removedNot stored, only delivered to subscribers
Late subscribersCan read older messages still in the listMiss messages sent before subscription
Typical use caseBackground jobs, task queuesReal-time notifications, chat updates
Reliability after restartMessages can remain if Redis persists dataMessages are lost on restart

For background processing, you usually want queues with lists, not Pub/Sub, because:

Pub/Sub is more for real-time events where it is acceptable to miss some messages.

Redis as a Broker for Celery

Instead of manually implementing your queue logic, you can use Celery, a popular task queue library in Python. Celery supports Redis as a broker and handles:

Here is a minimal setup.

Installing Celery with Redis

bash 
pip install celery[redis]

Defining a Celery app using Redis as broker

celery_app.py:

python 
from celery import Celery
celery_app = Celery(
    "tasks",
    broker="redis://localhost:6379/0",      # Redis as broker
    backend="redis://localhost:6379/1",     # Optional, store results
)

Declaring a task

tasks.py:

python 
import time
from celery_app import celery_app
@celery_app.task
def add(x, y):
    time.sleep(2)  # simulate long work
    return x + y

Sending a task from your web app

python 
from tasks import add
# Enqueue a task
result = add.delay(4, 5)  # returns immediately
print(result.id)  # task ID

Celery will push a message into Redis that looks something like:

Running a Celery worker

In a terminal:

bash 
celery -A tasks worker --loglevel=info

The worker connects to Redis, listens to the queue, pulls messages, executes them, and updates the result.

Getting the result later

python 
from tasks import add
result = add.delay(4, 5)
# Wait for the result (blocking)
value = result.get(timeout=10)
print(value)  # 9

Internally:

You do not need to manage Redis commands directly. Celery uses Redis lists and keys to implement a reliable queue.

Redis as a Broker with FastAPI

Let us combine FastAPI, Celery, and Redis. In a real project you would split files, but here is a simple example.

Project structure

text 
.
├─ celery_app.py
├─ tasks.py
└─ main.py

celery_app.py:

python 
from celery import Celery
celery_app = Celery(
    "background_tasks",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
)

tasks.py:

python 
import time
from celery_app import celery_app
@celery_app.task
def send_email(to: str, subject: str, body: str) -> str:
    time.sleep(5)  # simulate long operation
    # Here you would call an email provider
    print(f"Sending email to {to} with subject '{subject}'")
    return "ok"

main.py:

python 
from fastapi import FastAPI
from pydantic import BaseModel
from tasks import send_email
app = FastAPI()
class EmailRequest(BaseModel):
    to: str
    subject: str
    body: str
@app.post("/send-email")
def enqueue_email(email: EmailRequest):
    task = send_email.delay(email.to, email.subject, email.body)
    return {"task_id": task.id, "status": "queued"}

Running everything

  1. Start Redis:
bash 
   redis-server
  1. Start Celery worker:
bash 
   celery -A tasks worker --loglevel=info
  1. Start FastAPI:
bash 
   uvicorn main:app --reload
  1. Send a request:
bash 
   curl -X POST http://localhost:8000/send-email \
     -H "Content-Type: application/json" \
     -d '{"to": "alice@example.com", "subject": "Hi", "body": "Hello"}'

Response example:

json 
{
  "task_id": "a203f18f-0f34-4abd-b9b2-53d43b96a0be",
  "status": "queued"
}

The HTTP response is immediate, while the actual email task is processed in the background by Celery workers, with Redis acting as the message broker.

Designing Queues and Routing with Redis

In real systems, you often need multiple queues for different job types and priorities.

Multiple queues

Example queues:

Producers push to different lists:

python 
r.lpush("queue:emails", email_job_json)
r.lpush("queue:reports", report_job_json)

Workers can:

python 
# Prefer high_priority, fall back to low_priority
queue, raw_job = r.blpop("queue:high_priority", "queue:low_priority", timeout=0)

With Celery, you can define named queues and route tasks to them using configuration, but Redis is still the underlying broker that stores these per-queue lists.

Priority queues

Basic priority pattern with multiple lists:

Worker:

python 
while True:
    item = r.blpop("queue:high", "queue:medium", "queue:low", timeout=0)
    queue_name, raw_job = item
    # Process job

For simple systems, this is enough. For more complex priority rules you might use sorted sets with a score field, but that is more advanced.

Handling Retries and Failures

Background jobs fail for many reasons:

You should have a strategy for retries.

Manual retry strategy with Redis

A job may have fields like:

json 
{
  "id": 123,
  "task": "send_email",
  "payload": {
    "to": "alice@example.com"
  },
  "retry_count": 0,
  "max_retries": 5
}

Simplified worker logic:

python 
import json
import time
import redis
r = redis.Redis()
MAX_RETRIES = 5
while True:
    _, raw_job = r.blpop("queue:pending", timeout=0)
    job = json.loads(raw_job)
    try:
        # process job
        print(f"Sending email to {job['payload']['to']}")
        # ...
    except Exception as e:
        job["retry_count"] += 1
        if job["retry_count"] <= job["max_retries"]:
            # Requeue with some delay (very naive)
            time.sleep(2 ** job["retry_count"])  # exponential backoff
            r.lpush("queue:pending", json.dumps(job))
        else:
            # Move to dead-letter queue
            r.lpush("queue:failed", json.dumps(job))

This demonstrates:

Rule: Always have a way to inspect failed jobs, usually a dead-letter queue or a dedicated set or list like queue:failed. Never silently drop failures.

Visibility Timeouts and Acknowledgements

More advanced message brokers, such as AWS SQS, have a visibility timeout: after a worker takes a message, the message becomes invisible for a time. If the worker does not acknowledge it before the timeout, the message becomes visible again and can be processed by another worker.

Redis does not have this built in, but some Redis-based queues implement a similar idea:

Systems like RQ, Celery, and BullMQ (Node.js) implement acknowledgement and timeouts on top of Redis, so you do not have to.

When you choose Redis as a broker through such libraries, you automatically get:

Scaling Workers with Redis

One of the strengths of Redis as a broker is that you can add many workers, even on multiple machines, all pointing to the same Redis instance.

Horizontal scaling

Suppose you have heavy tasks:

You can run multiple worker processes or containers:

bash 
celery -A tasks worker --loglevel=info --concurrency=4
celery -A tasks worker --loglevel=info --concurrency=4

With 2 worker processes and concurrency of 4 each, you can run up to 8 tasks in parallel, all reading from the same queue in Redis.

Because Redis operations are atomic, only one worker gets each message.

Load balancing

Redis does not "load balance" by itself through any special feature. Instead, all workers pull from the same queue. The number of workers and their concurrency levels determine how much total work can be processed.

If you see:

You can:

Redis Broker Configuration and Tuning Basics

For real systems, you need to think about:

Persistence and durability

Redis is in memory, but can persist to disk using:

For a message broker, losing messages on a restart can be very bad.

Basic configuration options in redis.conf to enable AOF:

text 
appendonly yes
appendfsync everysec

This makes Redis slower but safer.

Eviction policies

If Redis runs out of memory, the eviction policy decides what to remove. For a broker, you usually do not want Redis to start evicting arbitrary keys that might be queue messages.

Example safe-ish choice in redis.conf for a broker:

text 
maxmemory-policy noeviction

With this, Redis will refuse writes if it is out of memory, instead of deleting messages. Your application then needs to handle errors.

Security basics

For production:

In redis.conf:

text 
  requirepass your_strong_password_here

In Python:

python 
  r = redis.Redis(host="redis-host", port=6379, password="your_strong_password_here")

Summary

Redis as a message broker gives you:

For backend development, Redis is often the easiest way to get from "I have a FastAPI app" to "I have background workers that process tasks reliably."

Views: 9

Comments

Please login to add a comment.

Don't have an account? Register now!

© 2025 Kahibaro
Visitors: 3505 / 24h
Impressum / Imprint
Datenschutz / Data Policy