Chapter 58: Docker Projects Part 2: AWS RDS, DynamoDB, S3, Reel Upload, Microservices, Queues, Logging, Monitoring and Cloud Deploys

58.8 Project 24: Worker queue with Celery and Redis, then RabbitMQ, with the Flower UI

By Ravindra Bagale · Last updated: 25 October 2026

Why: the problem

Suppose we work at Swiggy on the invoice team. After each order, the app must make a PDF-style invoice, send an email and update a report. Each step takes 2 to 5 seconds. If the web API does all of it before replying, the customer waits 10 seconds after "Place order". Many will tap again and create double orders. So the API replies at once and hands the slow work to background workers. In Python, the most used tool for this is Celery. Celery needs a broker, a place where tasks wait. We start with Redis because it is simple. Then we switch to RabbitMQ, a real message broker with acknowledgements, used by many banks and big shops. Flower is a web page that shows workers, tasks, failures and retries. We run all of it with Docker Compose and see the same code work with both brokers by changing one line.

What: every term in plain English

• Task — A Python function that Celery runs in the background, marked with @app.task.
• Broker — Where tasks wait: Redis or RabbitMQ.
• Result backend — Where results and status are saved, here Redis.
• Worker — A process (container) that takes tasks from the broker and runs them.
• Concurrency — How many tasks one worker runs at the same time, set with -c.
• Retry — Running a failed task again after a wait, for example when an email server is down.
• acks_late — The worker confirms a task only after it finishes, so a crash puts it back in the queue.
• Flower — A web dashboard for Celery on port 5555.

How: step by step

1. Write tasks. Three tasks: make invoice, send email (with retry), daily report.
2. Write the API. FastAPI that queues tasks and returns a task id.
3. Run with Redis. Compose: redis, api, worker, flower.
4. Watch in Flower. See tasks move from received to success or retry.
5. Switch to RabbitMQ. Change the broker URL, add the rabbitmq service, nothing else.
6. Scale and clean up. More workers with --scale, then down -v.

Architecture

Folder layout

jobs/
  compose.yaml          (redis, api, worker, email-worker, beat, flower)
  compose.rabbit.yaml   (adds rabbitmq)
  .env.rabbit
  app/celery_app.py  app/tasks.py  app/api.py
  app/requirements.txt  app/Dockerfile
  ops/load.sh  ops/cleanup.sh

Example 1: The Celery app and tasks, plus the API that queues them

Goal. The Celery app and tasks, plus the API that queues them.

app/celery_app.py

import os
from celery import Celery

BROKER = os.getenv("BROKER_URL", "redis://redis:6379/0")
BACKEND = os.getenv("RESULT_URL", "redis://redis:6379/1")

app = Celery("jobs", broker=BROKER, backend=BACKEND, include=["tasks"])
app.conf.update(
    task_acks_late=True,
    worker_prefetch_multiplier=1,
    task_reject_on_worker_lost=True,
    result_expires=3600,
    task_routes={"tasks.send_email": {"queue": "email"}},
    timezone="Asia/Kolkata",
    beat_schedule={
        "daily-report": {"task": "tasks.daily_report", "schedule": 300.0},
    },
)

app/tasks.py

import os, time, random, socket
from celery_app import app

ME = socket.gethostname()

@app.task(bind=True)
def make_invoice(self, order_id, amount):
    time.sleep(3)
    path = f"/data/invoices/INV-{order_id}.txt"
    os.makedirs("/data/invoices", exist_ok=True)
    with open(path, "w") as f:
        f.write(f"Invoice for order {order_id}: Rs {amount}")
    send_email.delay(order_id, path)
    return {"order": order_id, "file": path, "worker": ME}

@app.task(bind=True, autoretry_for=(ConnectionError,), retry_backoff=2, retry_kwargs={"max_retries": 4})
def send_email(self, order_id, path):
    if random.random() > 0.6:
        raise ConnectionError("mail server busy")
    time.sleep(1)
    return {"order": order_id, "sent": True, "try": self.request.retries + 1, "worker": ME}

@app.task
def daily_report():
    count = len(os.listdir("/data/invoices")) if os.path.isdir("/data/invoices") else 0
    return {"invoices_so_far": count}

app/api.py

from fastapi import FastAPI
from celery.result import AsyncResult
from celery_app import app as celery
from tasks import make_invoice

api = FastAPI()

@api.post("/orders/{order_id}", status_code=202)
def place(order_id: int, amount: float = 249.0):
    t = make_invoice.delay(order_id, amount)
    return {"order": order_id, "task_id": t.id}

@api.get("/tasks/{task_id}")
def status(task_id: str):
    r = AsyncResult(task_id, app=celery)
    return {"id": task_id, "state": r.state, "result": r.result if r.successful() else None}

app/requirements.txt

celery[redis]==5.4.0
flower==2.0.1
fastapi==0.115.2
uvicorn==0.32.0

app/Dockerfile

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY *.py .
RUN useradd -r -m app && mkdir -p /data && chown app /data
USER app
CMD ["celery", "-A", "celery_app", "worker", "-l", "info"]

What each line does

• include=['tasks'] — The worker loads tasks.py at start so it knows every task name.
• task_acks_late + reject_on_worker_lost — If a worker dies mid-task, the task goes back to the queue.
• prefetch_multiplier=1 — Each worker takes one task at a time. Long tasks do not pile up on one worker.
• task_routes — Emails go to their own queue email, so a slow mail server does not block invoices.
• autoretry_for + retry_backoff — On ConnectionError Celery retries after 2, 4, 8, 16 seconds, max 4 times.
• random failure — We fake a busy mail server 40 percent of the time to see retries in Flower.
• One image for all — api, worker, beat and flower use the same image with different commands.

Green output

$ docker build -t jobs-app ./app
 => naming to docker.io/library/jobs-app

Expected result. One image holds the API, the tasks and Flower.

Example 2: Compose with Redis as broker: api, two kinds of workers, beat scheduler and Flower

Goal. Compose with Redis as broker: api, two kinds of workers, beat scheduler and Flower.

compose.yaml

x-app: &app
  build: ./app
  image: jobs-app:1.0
  environment:
    BROKER_URL: ${BROKER_URL:-redis://redis:6379/0}
    RESULT_URL: redis://redis:6379/1
  volumes: [invoices:/data]
  depends_on:
    redis: { condition: service_healthy }
  restart: unless-stopped

services:
  redis:
    image: redis:7.4-alpine
    volumes: [redis-data:/data]
    healthcheck: { test: ["CMD", "redis-cli", "ping"], interval: 5s, retries: 10 }

  api:
    <<: *app
    command: ["uvicorn", "api:api", "--host", "0.0.0.0", "--port", "8000"]
    ports: ["8000:8000"]

  worker:
    <<: *app
    command: ["celery", "-A", "celery_app", "worker", "-Q", "celery", "-c", "2", "-l", "info"]
    stop_grace_period: 60s

  email-worker:
    <<: *app
    command: ["celery", "-A", "celery_app", "worker", "-Q", "email", "-c", "4", "-l", "info"]

  beat:
    <<: *app
    command: ["celery", "-A", "celery_app", "beat", "-l", "info", "-s", "/tmp/beat-schedule"]

  flower:
    <<: *app
    command: ["celery", "-A", "celery_app", "flower", "--port=5555", "--basic_auth=admin:change-me-flower"]
    ports: ["5555:5555"]

volumes:
  redis-data:
  invoices:

What each line does

• BROKER_URL default — Uses Redis unless we set BROKER_URL in .env. This is how we switch to RabbitMQ later.
• -Q celery / -Q email — Two worker groups listen to two queues.
• -c 2 and -c 4 — Invoice workers run 2 tasks at once, email workers 4.
• beat — Only one beat container. It sends daily_report every 300 seconds for this demo.
• stop_grace_period 60s — docker stop waits for running invoices to finish.
• --basic_auth — Flower has a login. Change the placeholder password.

Green output

$ docker compose up -d --build
 Container jobs-redis-1  Healthy
 Container jobs-worker-1  Started
 Container jobs-email-worker-1  Started
 Container jobs-flower-1  Started
$ for i in 101 102 103 104 105; do curl -s -X POST localhost:8000/orders/$i; echo; done
{"order":101,"task_id":"3f1c8a0e-5b2d-4c9e-9a11-6f0d7e2b4c55"}
...
$ docker compose logs email-worker | grep -E 'retry|succeeded' | head -3
email-worker-1  | Task tasks.send_email[9b2e...] retry: Retry in 2s: ConnectionError('mail server busy')
email-worker-1  | Task tasks.send_email[9b2e...] succeeded in 1.0s: {'order': 102, 'sent': True, 'try': 2, ...}

Expected result. The API answered at once. Workers made invoices, and failed emails were retried and then sent.

Example 3: Switch the broker to RabbitMQ with one line, look at queues, and scale workers

Goal. Switch the broker to RabbitMQ with one line, look at queues, and scale workers.

compose.rabbit.yaml

services:
  rabbitmq:
    image: rabbitmq:3.13-management-alpine
    environment:
      RABBITMQ_DEFAULT_USER: jobs
      RABBITMQ_DEFAULT_PASS: change-me-rabbit
    ports: ["15672:15672"]
    volumes: [rabbit-data:/var/lib/rabbitmq]
    healthcheck:
      test: ["CMD", "rabbitmq-diagnostics", "-q", "ping"]
      interval: 10s
      retries: 10

  api:
    depends_on: { rabbitmq: { condition: service_healthy } }
  worker:
    depends_on: { rabbitmq: { condition: service_healthy } }
  email-worker:
    depends_on: { rabbitmq: { condition: service_healthy } }

volumes:
  rabbit-data:

.env.rabbit

BROKER_URL=amqp://jobs:change-me-rabbit@rabbitmq:5672//

ops/load.sh

for i in $(seq 200 260); do
  curl -s -o /dev/null -X POST "localhost:8000/orders/$i"
done
echo "61 orders queued"

ops/cleanup.sh

docker compose -f compose.yaml -f compose.rabbit.yaml --env-file .env.rabbit down -v --rmi local
docker compose down -v --rmi local 2>/dev/null
echo "jobs stack removed"

What each line does

• -f compose.yaml -f compose.rabbit.yaml — Compose merges the two files. RabbitMQ is added, and the workers wait for it.
• amqp://...// — The last / is the virtual host. The default vhost is /, so the URL ends with //.
• Redis stays — Redis is still the result backend. Only the broker changes.
• management image — Port 15672 has a web UI that shows queues celery and email and message rates.
• Durable queues — RabbitMQ keeps queues and messages on the rabbit-data volume across restarts.
• --scale worker=4 — Four invoice worker containers share the celery queue fairly because prefetch is 1.

Green output

$ docker compose down
$ docker compose -f compose.yaml -f compose.rabbit.yaml --env-file .env.rabbit up -d
 Container jobs-rabbitmq-1  Healthy
$ docker compose -f compose.yaml -f compose.rabbit.yaml --env-file .env.rabbit exec rabbitmq rabbitmqctl list_queues name messages consumers
name     messages  consumers
celery   0         1
email    0         1
$ bash ops/load.sh
61 orders queued
$ docker compose -f compose.yaml -f compose.rabbit.yaml --env-file .env.rabbit up -d --scale worker=4
$ docker compose -f compose.yaml -f compose.rabbit.yaml --env-file .env.rabbit exec rabbitmq rabbitmqctl list_queues name messages consumers
celery   37        4
email    5         1

Expected result. Same code, new broker. With 4 workers the celery queue drains about twice as fast.

Every command, step by step (plain docker and Compose)

1. Build and start

   docker compose up -d --build
   

   What it does: Redis, API, workers, beat and Flower.

   Container jobs-flower-1 Started
   

2. Place an order

   curl -s -X POST localhost:8000/orders/101
   

   What it does: Returns a task id at once.

   {"order":101,"task_id":"3f1c..."}
   

3. Check the task

   curl -s localhost:8000/tasks/TASK_ID
   

   What it does: PENDING, then SUCCESS after about 3 seconds.

   {"state":"SUCCESS",...}
   

4. See the invoice file

   docker compose exec worker ls /data/invoices
   

   What it does: Files on the shared volume.

   INV-101.txt
   

5. Open Flower

   http://localhost:5555 (admin / change-me-flower)
   

   What it does: Workers, tasks, success and retry counts.

   2 workers online
   

6. Watch retries

   docker compose logs -f email-worker
   

   What it does: Some emails retry and then succeed.

   retry: Retry in 2s
   

7. Check beat

   docker compose logs beat | tail -2
   

   What it does: daily_report is sent on schedule.

   Scheduler: Sending due task daily-report
   

8. Switch to RabbitMQ

   docker compose down && docker compose -f compose.yaml -f compose.rabbit.yaml --env-file .env.rabbit up -d
   

   What it does: Same code, new broker.

   Container jobs-rabbitmq-1 Healthy
   

9. Open RabbitMQ UI

   http://localhost:15672 (jobs / change-me-rabbit)
   

   What it does: Queues celery and email.

   Queues: 2
   

10. Load test

    bash ops/load.sh
    

    What it does: 61 orders queued.

    61 orders queued
    

11. Scale workers

    docker compose -f compose.yaml -f compose.rabbit.yaml --env-file .env.rabbit up -d --scale worker=4
    

    What it does: Queue drains faster.

    Container jobs-worker-4 Started
    

12. Clean up

    bash ops/cleanup.sh
    

    What it does: Removes containers, volumes and image.

    jobs stack removed
    

When it is red 1

celery.exceptions.NotRegistered: 'tasks.make_invoice'

Cause. The worker did not load tasks.py.

Fix, step by step

1. Keep include=['tasks'] in the Celery app.
2. Rebuild the image after adding a task.

When it is red 2

consumer: Cannot connect to redis://redis:6379/0: Error -2 connecting to redis:6379. Name or service not known.

Cause. The worker started on a different network or before Redis was ready.

Fix, step by step

1. Keep depends_on with service_healthy.
2. Use the service name redis, not localhost.

When it is red 3

amqp ACCESS_REFUSED - Login was refused using authentication mechanism PLAIN

Cause. Wrong RabbitMQ user or password, or the volume was made with old credentials.

Fix, step by step

1. Match BROKER_URL with RABBITMQ_DEFAULT_USER and PASS.
2. For a test stack: down -v and start again.

When it is red 4

Tasks stay PENDING forever

Cause. No worker listens on that queue, or the task id is wrong. PENDING also means unknown.

Fix, step by step

1. Check -Q names match task_routes.
2. Look in Flower for the task.

When it is red 5

beat: Permission denied: 'celerybeat-schedule'

Cause. beat writes its schedule file in /app, which the non-root user cannot write.

Fix, step by step

1. Use -s /tmp/beat-schedule as shown.

When it is red 6

WorkerLostError: Worker exited prematurely: signal 9 (SIGKILL)

Cause. The worker container ran out of memory, or docker stop killed it after the grace period.

Fix, step by step

1. Check docker stats and raise the memory limit.
2. Raise stop_grace_period so long tasks can finish.
3. Thanks to acks_late the task goes back to the queue.

When it is red 7

Two daily reports every time

Cause. Two beat containers are running.

Fix, step by step

1. Run exactly one beat. Never scale beat.

Practice tasks

docker compose up -d --build

curl -s localhost:8000/tasks/ID

http://localhost:5555

docker compose exec redis redis-cli LLEN celery

docker kill jobs-worker-1

rabbitmqctl list_queues

--scale worker=4

link_error= or on_failure

task_routes

bash ops/cleanup.sh

Recap

• Background workers keep APIs fast; the API returns 202 and a task id.
• Celery tasks wait in a broker: Redis is simple, RabbitMQ adds strong delivery features.
• acks_late and prefetch 1 make long tasks safe and fair.
• Separate queues stop slow jobs from blocking others.
• Use retries with backoff for flaky outside services.
• Run exactly one beat; scale workers with --scale; watch with Flower.

FAQ

Q. Redis or RabbitMQ for Celery?

Redis for simple setups. RabbitMQ when you need strong delivery, many queues and routing.

Q. Can I use Amazon SQS?

Yes, Celery supports SQS as a broker. Results then go to Redis or a database.

Q. Is Celery only for Python?

Yes. Node.js has BullMQ, Java has Spring with RabbitMQ, and the broker ideas are the same.

Q. How do I see failed tasks?

Flower shows them, and the result backend keeps the error and traceback.

Q. What is a chord or a chain?

Ways to join tasks. chain runs tasks one after another; group runs many at once; chord runs a group and then one final task with all results.

Q. How do I deploy workers on AWS?

Run the same image on ECS or EC2 with Amazon MQ (RabbitMQ) or ElastiCache (Redis) as the broker. Only the URLs change.

Q. How do I test tasks without a broker?

Set task_always_eager=True in tests; tasks then run inside the test process.

Q. Should Flower be public?

No. Keep basic auth and put it behind a VPN or Traefik with IP allow lists.

Q. How many workers do I need?

Watch queue length. If it keeps growing, add workers or raise concurrency.

Interview questions

1. What is Celery and why use it?

   Model answer: A Python task queue for running slow or scheduled work in background workers.

2. What is the difference between broker and result backend?

   Model answer: The broker holds tasks waiting; the backend stores task state and results.

3. How do you avoid losing tasks if a worker crashes?

   Model answer: acks_late with reject_on_worker_lost, durable queues, and idempotent tasks.

4. How do you scale Celery in Docker?

   Model answer: Run more worker containers with --scale or replicas, separated by queue.

5. What does idempotent mean for a task?

   Model answer: Running it twice gives the same result, so retries are safe.

Samajla ka?

Ravi Sir's Cards for this lesson

Revision cardInterview cardComparison card

मराठी cards