Skip to content

Mini Project: RabbitMQ Notification System

DodaTech Updated 2026-06-28 6 min read

In this tutorial, you will learn about Mini Project: RabbitMQ Notification System. We cover key concepts, practical examples, and best practices to help you master this topic.

Build a complete notification system with RabbitMQ using topic exchanges, multiple consumer types, TLS security, and publisher confirms for reliable delivery.

What You Learn

You will build a production-ready notification system that routes different notification types to different handlers, implements reliable delivery with publisher confirms, and secures all communication with TLS.

Why It Matters

A notification system is a common real-world messaging application. This project combines all RabbitMQ concepts you learned: exchanges, routing, acknowledgements, QoS, persistence, and security. Completing it proves you can build production messaging systems.

Real-World Use

This is the same pattern Doda Browser uses for notifications. When a malware scan completes, the system publishes a notification. Different consumers handle email, SMS, and push notifications. The topic exchange routes each type to the correct handler.

System Architecture

flowchart TB
    P[Notification Producer] --> TE[Topic Exchange
notifications]
    TE -->|"email.*"| EQ[Email Queue]
    TE -->|"sms.*"| SQ[SMS Queue]
    TE -->|"push.*"| PQ[Push Queue]
    TE -->|"alert.*"| AQ[Alert Queue]
    EQ --> EC[Email Consumer]
    SQ --> SC[SMS Consumer]
    PQ --> PC[Push Consumer]
    AQ --> AC[Alert Consumer]
    style TE fill:#f90,color:#fff

Setup

# requirements.txt
# pika==1.3.2
# requests==2.31.0

Step 1: Exchange and Queue Setup

import pika
import json
import os
import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')

RABBITMQ_HOST = os.environ.get('RABBITMQ_HOST', 'localhost')
RABBITMQ_PORT = int(os.environ.get('RABBITMQ_PORT', 5672))
EXCHANGE_NAME = 'notifications'
QUEUES = {
    'email_queue': 'email.#',
    'sms_queue': 'sms.#',
    'push_queue': 'push.#',
    'alert_queue': 'alert.#',
}

def setup_infrastructure():
    connection = pika.BlockingConnection(
        pika.ConnectionParameters(host=RABBITMQ_HOST, port=RABBITMQ_PORT)
    )
    channel = connection.channel()

    channel.exchange_declare(exchange=EXCHANGE_NAME, exchange_type='topic', durable=True)
    channel.confirm_delivery()

    for queue_name, binding_key in QUEUES.items():
        channel.queue_declare(queue=queue_name, durable=True)
        channel.queue_bind(exchange=EXCHANGE_NAME, queue=queue_name, routing_key=binding_key)
        logging.info(f"Queue {queue_name} bound with key {binding_key}")

    logging.info("Infrastructure setup complete")
    connection.close()

if __name__ == '__main__':
    setup_infrastructure()

Expected output:

2026-06-28 10:00:00 [INFO] Queue email_queue bound with key email.#
2026-06-28 10:00:00 [INFO] Queue sms_queue bound with key sms.#
2026-06-28 10:00:00 [INFO] Queue push_queue bound with key push.#
2026-06-28 10:00:00 [INFO] Queue alert_queue bound with key alert.#
2026-06-28 10:00:00 [INFO] Infrastructure setup complete

Step 2: Notification Producer

import pika
import json
import uuid
import logging

logger = logging.getLogger(__name__)

class NotificationProducer:
    def __init__(self):
        self.connection = pika.BlockingConnection(
            pika.ConnectionParameters(host=RABBITMQ_HOST, port=RABBITMQ_PORT)
        )
        self.channel = self.connection.channel()
        self.channel.confirm_delivery()
        logger.info("Notification producer initialized")

    def send_notification(self, notification_type, recipient, subject, body, priority='normal'):
        routing_key = f"{notification_type}.{priority}"
        message = {
            'id': str(uuid.uuid4()),
            'type': notification_type,
            'recipient': recipient,
            'subject': subject,
            'body': body,
            'priority': priority,
            'timestamp': '2026-06-28T10:00:00Z'
        }

        try:
            self.channel.basic_publish(
                exchange=EXCHANGE_NAME,
                routing_key=routing_key,
                body=json.dumps(message),
                properties=pika.BasicProperties(
                    delivery_mode=2,
                    content_type='application/json',
                    message_id=message['id']
                )
            )
            logger.info(f"Sent {notification_type} to {recipient}: {subject}")
            return True
        except (pika.exceptions.UnroutableError, pika.exceptions.NackError) as e:
            logger.error(f"Failed to send notification: {e}")
            return False

    def close(self):
        if self.connection and self.connection.is_open:
            self.connection.close()

if __name__ == '__main__':
    producer = NotificationProducer()
    producer.send_notification('email', 'user@example.com', 'Scan Complete', 'Your file is clean')
    producer.send_notification('sms', '+1234567890', 'Alert', 'Suspicious activity detected', 'high')
    producer.send_notification('push', 'device_token_123', 'Update Available', 'Version 2.1 is ready')
    producer.close()

Expected output:

2026-06-28 10:00:00 [INFO] Sent email to user@example.com: Scan Complete
2026-06-28 10:00:00 [INFO] Sent sms to +1234567890: Alert
2026-06-28 10:00:00 [INFO] Sent push to device_token_123: Update Available

Step 3: Notification Consumers

import pika
import json
import logging
import time
import threading

logger = logging.getLogger(__name__)

class BaseConsumer:
    def __init__(self, queue_name, handler_name):
        self.queue_name = queue_name
        self.handler_name = handler_name
        self.connection = pika.BlockingConnection(
            pika.ConnectionParameters(host=RABBITMQ_HOST, port=RABBITMQ_PORT)
        )
        self.channel = self.connection.channel()
        self.channel.basic_qos(prefetch_count=1)

    def process(self, notification):
        raise NotImplementedError

    def callback(self, ch, method, properties, body):
        notification = json.loads(body)
        try:
            self.process(notification)
            ch.basic_ack(delivery_tag=method.delivery_tag)
            logger.info(f"[{self.handler_name}] Processed: {notification['id']}")
        except Exception as e:
            logger.error(f"[{self.handler_name}] Error: {e}")
            ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

    def run(self):
        self.channel.basic_consume(
            queue=self.queue_name,
            on_message_callback=self.callback
        )
        logger.info(f"[{self.handler_name}] Waiting for messages on {self.queue_name}")
        self.channel.start_consuming()

class EmailConsumer(BaseConsumer):
    def __init__(self):
        super().__init__('email_queue', 'Email')

    def process(self, notification):
        logger.info(f"Sending email to {notification['recipient']}")
        logger.info(f"Subject: {notification['subject']}")
        logger.info(f"Body: {notification['body']}")
        time.sleep(0.3)

class SMSConsumer(BaseConsumer):
    def __init__(self):
        super().__init__('sms_queue', 'SMS')

    def process(self, notification):
        logger.info(f"Sending SMS to {notification['recipient']}")
        logger.info(f"Message: {notification['subject']}: {notification['body']}")
        time.sleep(0.2)

class PushConsumer(BaseConsumer):
    def __init__(self):
        super().__init__('push_queue', 'Push')

    def process(self, notification):
        logger.info(f"Sending push to device {notification['recipient']}")
        logger.info(f"Title: {notification['subject']}")
        time.sleep(0.1)

class AlertConsumer(BaseConsumer):
    def __init__(self):
        super().__init__('alert_queue', 'Alert')

    def process(self, notification):
        logger.warning(f"HIGH PRIORITY ALERT for {notification['recipient']}")
        logger.warning(f"Subject: {notification['subject']}")
        time.sleep(0.1)

if __name__ == '__main__':
    consumers = [
        EmailConsumer(),
        SMSConsumer(),
        PushConsumer(),
        AlertConsumer(),
    ]

    threads = []
    for consumer in consumers:
        t = threading.Thread(target=consumer.run, daemon=True)
        t.start()
        threads.append(t)

    logger.info("All consumers started")
    time.sleep(60)

Step 4: Monitoring Dashboard

import requests
import json
import time
import os

MONITOR_URL = os.environ.get('RABBITMQ_API_URL', 'http://localhost:15672/api')
AUTH = (os.environ.get('RABBITMQ_USER', 'guest'), os.environ.get('RABBITMQ_PASS', 'guest'))

def get_queue_metrics():
    queues = requests.get(f"{MONITOR_URL}/queues", auth=AUTH).json()
    metrics = {}
    for q in queues:
        if q['name'] in QUEUES:
            metrics[q['name']] = {
                'ready': q['messages_ready'],
                'unacked': q['messages_unacknowledged'],
                'total': q['messages'],
                'consumers': q['consumers'],
            }
    return metrics

def check_health():
    try:
        health = requests.get(f"{MONITOR_URL}/healthchecks/node", auth=AUTH, timeout=5)
        return health.status_code == 200
    except:
        return False

if __name__ == '__main__':
    print(f"Exchange: {EXCHANGE_NAME}")
    print("Monitoring notification system...")
    print()

    while True:
        metrics = get_queue_metrics()
        healthy = check_health()

        print(f"Time: {time.strftime('%H:%M:%S')} | Healthy: {healthy}")
        print("-" * 60)

        for queue_name, data in metrics.items():
            status = "OK" if data['ready'] < 100 else "WARN"
            print(f"  {queue_name:20s} | Ready: {data['ready']:4d} | "
                  f"Unacked: {data['unacked']:3d} | "
                  f"Consumers: {data['consumers']:2d} | {status}")

        print()
        time.sleep(5)

Expected output:

Exchange: notifications
Time: 10:00:00 | Healthy: True
------------------------------------------------------------
  email_queue           | Ready:    0 | Unacked:   0 | Consumers: 1 | OK
  sms_queue             | Ready:    0 | Unacked:   0 | Consumers: 1 | OK
  push_queue            | Ready:    0 | Unacked:   0 | Consumers: 1 | OK
  alert_queue           | Ready:    0 | Unacked:   0 | Consumers: 1 | OK

Common Mistakes

1. Not Using Durable Queues

Without durable queues, all queue definitions and messages are lost on broker restart. Always set durable=True in production.

2. Mixing Up Routing Keys

Routing keys like email.high only match email.# or email.*, not email alone. Test routing keys with the management UI before deploying.

3. Forgetting to Set prefetch_count

Without basic_qos(prefetch_count=1), consumers buffer all messages, causing OOM on high-traffic queues.

4. Not Handling Consumer Exceptions

An unhandled exception in a consumer callback crashes the consumer. Wrap processing in try/except and nack on failure.

5. Running Without Monitoring

Queues can silently grow to millions of messages. Set up the monitoring dashboard with alerts for queue depth, consumer count, and message rates.

Practice Questions

1. Why use a topic exchange instead of direct exchange for this project?

Topic exchange supports pattern matching (email.#), allowing flexible routing. Direct exchange requires exact match. New notification types can be added without changing the producer.

2. What happens if a consumer crashes mid-processing?

The message remains unacknowledged. On connection loss, RabbitMQ requeues it for another consumer. The manual ack pattern ensures no message is lost.

3. How can you add SMS notification for high priority only?

Change the SMS binding key to sms.high instead of sms.#. Only messages with routing key sms.high reach the SMS queue.

4. Why is prefetch_count=1 important here?

Each consumer handles one notification at a time. If an email takes 300ms and a push takes 100ms, prefetch=1 ensures fair distribution without any consumer being overwhelmed.

Challenge

Extend the notification system: add a dead letter exchange for failed notifications, implement retry with exponential backoff (3 attempts), and route retries to a separate delay queue using TTL. The system should log every failed notification to a dead letter queue for manual inspection.

FAQ

Can this system handle millions of notifications per day?

Yes. RabbitMQ processes thousands of messages per second. With consumers in separate processes and horizontal scaling, millions per day is achievable.

How do I deploy this to production?

Package each consumer as a separate service. Use Docker containers. Run multiple consumer instances behind a load balancer connecting to a RabbitMQ cluster.

How do I handle notification delivery failures?

Use dead letter exchanges for failed messages. Implement consumer retry logic with exponential backoff. Log all failed deliveries for manual review.

Can I add new notification types without downtime?

Yes. Add a new queue and binding to the topic exchange. New consumers can start consuming without affecting existing ones.

How do I ensure notifications are not lost during broker restart?

Use durable queues, persistent messages (delivery_mode=2), and publisher confirms. In a cluster with quorum queues, messages survive node failures.

What's Next

You completed the RabbitMQ notification system. Review message queue patterns to see how this fits into the broader messaging landscape, or explore Celery for Python task queues built on RabbitMQ.

← Previous RabbitMQ Security — Complete Guide Next → RabbitMQ Architecture Deep Dive -- Broker Internals Explained

Built by the developers of DodaTech

Doda Browser, DodaZIP & Durga Antivirus Pro

Home Browse Rabbitmq