254-Day Hands-On System Design Series
Module 2: Scalable Log Processing | Week 5: Message Queues

Picture a busy postal sorting facility. Letters arrive constantly, and workers need to route them to different destinations based on addresses, priority levels, or content types. Your distributed log processing system faces the same challenge—millions of log entries need intelligent routing to appropriate processing pipelines.

Today, we'll implement different exchange types that act as intelligent routing engines, directing logs based on topics, sources, and severity levels. By lesson's end, you'll have a robust routing system that automatically sends database logs to analytics pipelines while routing security alerts to monitoring systems.

🎯 Learning Outcome: Build a production-ready routing system handling 1000+ messages/second with multiple exchange patterns

The Routing Challenge

Yesterday, we built reliable message processing with acknowledgments. Now we need sophisticated routing. Consider Netflix's logging infrastructure—user interaction logs go to recommendation engines, performance metrics flow to monitoring dashboards, and error logs route to debugging systems. Each requires different processing approaches.

Subscribe Now

GitHub Link:

https://github.com/sysdr/course-p/tree/main/day35/day35-exchange-routing

Traditional point-to-point messaging creates rigid connections. Instead, we need flexible routing patterns that adapt as your system grows. Exchange types solve this by providing standardized routing mechanisms.

Understanding Exchange Types

Direct Exchange: Precise Routing

Direct exchanges route messages using exact routing key matches. Think of them as dedicated mail slots—each key opens exactly one destination.

# Route database logs to specific handler
routing_key = "database.postgres.error"
# Only consumers bound to this exact key receive messages

Topic Exchange: Pattern-Based Intelligence

Topic exchanges use wildcard patterns for flexible routing. They're like smart postal workers who understand addressing patterns.

# Route all database logs regardless of specific database
pattern = "database.*.error"
# Matches: database.postgres.error, database.mysql.error

Fanout Exchange: Broadcast Distribution

Fanout exchanges send copies to all bound queues, like announcing over a PA system.

# Critical security alerts go to ALL monitoring systems
exchange_type = "fanout"
# Every bound queue receives the message

Architecture Deep Dive

Our routing system consists of four key components:

Message Producer: Generates logs with structured routing keys following hierarchical patterns (service.component.level).

Exchange Router: Evaluates routing keys against bound patterns, determining destination queues for each message.

Queue Bindings: Define relationships between exchanges and queues using routing patterns or exact matches.

Specialized Consumers: Process messages from specific queues, each optimized for particular log types.

The data flow follows a clear pattern: producers publish with routing keys, exchanges evaluate against bindings, messages route to matching queues, and specialized consumers process based on log types.

Implementation Strategy

We'll build a Python-based routing system using RabbitMQ's exchange types. The implementation includes:

Log Message Structure: Standardized format with timestamp, service, component, level, and payload fields.

Exchange Configuration: Programmatic setup of different exchange types with appropriate bindings.

Smart Producers: Generate realistic log messages with proper routing keys based on service architecture.

Processing Pipelines: Specialized consumers that simulate real-world log processing (analytics, monitoring, alerting).

The routing key format follows {service}.{component}.{level} convention, enabling both precise and pattern-based routing.

🛠️ Hands-On Implementation

Project Structure Setup

Create the complete project structure:

mkdir -p day35-exchange-routing/{src,tests,config,logs,web,scripts}
cd day35-exchange-routing

Core Dependencies

# requirements.txt
pika==1.3.2
flask==3.0.3
pytest==8.2.0
redis==5.0.4
flask-socketio==5.3.6
colorama==0.4.6
requests==2.31.0

Exchange Manager Implementation

The heart of our routing system:

# src/exchange_manager.py
class ExchangeManager:
    def __init__(self):
        self.connection = None
        self.channel = None
        self.setup_logging()
        
    def setup_logging(self):
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s'
        )
        self.logger = logging.getLogger(__name__)
        
    def connect(self):
        """Establish connection to RabbitMQ"""
        try:
            credentials = pika.PlainCredentials(
                Config.RABBITMQ_USER, 
                Config.RABBITMQ_PASSWORD
            )
            parameters = pika.ConnectionParameters(
                host=Config.RABBITMQ_HOST,
                port=Config.RABBITMQ_PORT,
                virtual_host=Config.RABBITMQ_VHOST,
                credentials=credentials
            )
            self.connection = pika.BlockingConnection(parameters)
            self.channel = self.connection.channel()
            self.logger.info("Connected to RabbitMQ")
            return True
        except Exception as e:
            self.logger.error(f"Failed to connect to RabbitMQ: {e}")
            return False
......

Smart Log Producer

Generates realistic log patterns:

# src/log_producer.py
class LogProducer:
    def __init__(self):
        self.connection = None
        self.channel = None
        
    def connect(self):
        """Connect to RabbitMQ"""
        credentials = pika.PlainCredentials(Config.RABBITMQ_USER, Config.RABBITMQ_PASSWORD)
        parameters = pika.ConnectionParameters(
            host=Config.RABBITMQ_HOST,
            port=Config.RABBITMQ_PORT,
            credentials=credentials
        )
        self.connection = pika.BlockingConnection(parameters)
        self.channel = self.connection.channel()

Specialized Consumers

Process messages by type:

# src/log_consumer.py  

class LogConsumer:
    def __init__(self, queue_name, consumer_id):
        self.queue_name = queue_name
        self.consumer_id = consumer_id
        self.connection = None
        self.channel = None
        self.message_count = 0
        
    def connect(self):
        """Connect to RabbitMQ"""
        credentials = pika.PlainCredentials(Config.RABBITMQ_USER, Config.RABBITMQ_PASSWORD)
        parameters = pika.ConnectionParameters(
            host=Config.RABBITMQ_HOST,
            port=Config.RABBITMQ_PORT,
            credentials=credentials
        )
        self.connection = pika.BlockingConnection(parameters)
        self.channel = self.connection.channel()
        
    def process_message(self, ch, method, properties, body):
        """Process incoming log message"""
        try:
            message = json.loads(body)
            self.message_count += 1
            
            print(f"\n🔍 [{self.consumer_id}] Processing message #{self.message_count}")
            print(f"    Queue: {self.queue_name}")
            print(f"    Service: {message.get('service', 'unknown')}")
            print(f"    Component: {message.get('component', 'unknown')}")
            print(f"    Level: {message.get('level', 'unknown')}")
            print(f"    Message: {message.get('message', '')}")
            print(f"    Timestamp: {message.get('timestamp', '')}")
            
            # Simulate processing based on message type
            if message.get('level') == 'error':
                print("    🚨 ERROR PROCESSING: Sending to incident management")
            elif message.get('service') == 'security':
                print("    🔒 SECURITY PROCESSING: Analyzing for threats")
            elif message.get('service') == 'database':
                print("    💾 DATABASE PROCESSING: Performance analysis")
            else:
                print("    ✅ STANDARD PROCESSING: Logged and indexed")
                
            # Acknowledge message
            ch.basic_ack(delivery_tag=method.delivery_tag)
            
        except Exception as e:
            print(f"❌ Error processing message: {e}")
            # Reject and requeue
            ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
            
    def start_consuming(self):
        """Start consuming messages"""
        self.channel.basic_qos(prefetch_count=1)
        self.channel.basic_consume(
            queue=self.queue_name,
            on_message_callback=self.process_message
        )
        
        print(f"🚀 [{self.consumer_id}] Starting to consume from {self.queue_name}")
        print("Press CTRL+C to stop...")
        
        try:
            self.channel.start_consuming()
        except KeyboardInterrupt:
            self.channel.stop_consuming()

Real-Time Web Dashboard

Monitor routing in action:

# web/dashboard.py
    
    def connect_and_monitor(self):
        """Monitor all queues and emit stats with reconnection logic"""
        while self.should_reconnect:
            try:
                if not self.connect():
                    continue
                    
                def process_dashboard_message(ch, method, properties, body):
                    try:
                        message = json.loads(body)
                        self.update_stats(message, method.routing_key)
                        socketio.emit('log_update', {
                            'message': message,
                            'stats': self.stats,
                            'queue': method.routing_key
                        })
                        ch.basic_ack(delivery_tag=method.delivery_tag)
                    except Exception as e:
                        print(f"Dashboard processing error: {e}")
                        
                # Monitor all queues
                for queue_name in Config.QUEUES.values():
                    self.channel.basic_consume(
                        queue=queue_name,
                        on_message_callback=process_dashboard_message
                    )
                    
                print("✅ Started monitoring queues")
                self.channel.start_consuming()
                
            except pika.exceptions.AMQPConnectionError:
                print("❌ Lost connection to RabbitMQ")
                if self.connection and not self.connection.is_closed:
                    self.connection.close()
                time.sleep(5)  # Wait before reconnecting
            except Exception as e:
                print(f"❌ Dashboard monitoring error: {e}")
                time.sleep(5)  # Wait before retrying

🧪 Build, Test & Verify Guide

Quick Setup Commands

# Install dependencies
pip install -r requirements.txt

# Run comprehensive tests
python -m pytest tests/ -v

# Start infrastructure
docker-compose up -d

# Run demonstration
python run_system.py

Docker Configuration

# docker-compose.yml
services:
  rabbitmq:
    image: rabbitmq:3.12-management
    ports:
      - "5672:5672"
      - "15672:15672"
    environment:
      RABBITMQ_DEFAULT_USER: guest
      RABBITMQ_DEFAULT_PASS: guest

Testing Strategy

# tests/test_exchange_routing.py
def test_direct_exchange_routing(self):
    message = self.producer.create_log_message(
        'database', 'postgres', 'error', 'Test direct routing'
    )
    self.producer.publish_to_direct('database.postgres.error', message)
    assert message['routing_key'] == 'database.postgres.error'

def test_topic_exchange_patterns(self):
    test_cases = [
        ('database.postgres.info', 'database'),
        ('api.gateway.warning', 'api'),
        ('security.auth.error', 'security')
    ]
    # Verify pattern matching works correctly

Expected Results

✅ Successful Build Indicators:

All tests pass (12/12)
RabbitMQ running on localhost:5672
Web dashboard accessible at http://localhost:5000
Console showing routed messages by type
Real-time statistics updating

🔍 Demo Output:

🎯 Direct: database.postgres.error
🏷️  Topic: api.gateway.info  
📢 Fanout: Critical security message
✅ [DATABASE-PROCESSOR] Processing message #1
🚨 ERROR PROCESSING: Sending to incident management

Real-World Impact

Major platforms rely on sophisticated log routing. GitHub routes deployment logs to CI/CD dashboards while sending performance metrics to capacity planning systems. Slack separates user activity logs from infrastructure monitoring, ensuring each team receives relevant information without noise.

This routing flexibility becomes critical during incidents. Security teams need immediate access to authentication logs, while performance engineers focus on latency metrics. Proper routing ensures the right information reaches the right teams instantly.

Building Your Router

Start with a simple direct exchange for exact matching. Add topic exchanges for pattern-based routing. Implement fanout for critical broadcasts. Each exchange type serves specific use cases—direct for precise routing, topic for flexible patterns, fanout for wide distribution.

Configure queue bindings that reflect your actual service architecture. If you have user services, database services, and API gateways, create routing patterns that match this structure. Design routing keys that grow with your system.

Test with realistic log volumes. Start with hundreds of messages per second, then scale to thousands. Monitor routing performance and queue depths. Proper routing reduces downstream processing load by ensuring consumers only handle relevant messages.

📝 Assignment: E-Commerce Routing Challenge

Objective: Implement routing for a fictional e-commerce platform with user activity, inventory changes, and payment processing.

Requirements:

Design hierarchical routing keys for three services
Configure topic exchanges with wildcard bindings
Build specialized consumers for each service area
Test with 500+ messages demonstrating routing patterns
Measure routing performance and queue distribution

Solution Approach:

Service identification: user, inventory, payment
Routing keys: payment.processor.error, inventory.stock.update
Topic patterns: payment.*, inventory.*, user.activity.*
Specialized processing: payment errors → security, inventory → analytics
Performance testing: sustained 1000 msg/sec with <10ms routing latency

Success Metrics

Your routing system succeeds when messages reach appropriate consumers within milliseconds, regardless of volume. Queue depths remain stable under load. Adding new routing patterns requires no code changes to existing consumers. Failed routing attempts trigger clear error messages.

🔄 Tomorrow's Challenge

Next, we'll add dead letter queues for handling processing failures. When log processing fails repeatedly, these messages need special handling to prevent system degradation. You'll learn to build resilient error handling that maintains system stability during processing failures.

💡 Key Takeaway

Smart routing transforms chaotic log streams into organized processing pipelines. Master exchange types to build systems that scale efficiently and route intelligently, ensuring each log message reaches its optimal destination for maximum processing value.

🚀 Ready to Build?

Clone the complete implementation:

git clone https://github.com/systemdesign/day35-exchange-routing
cd day35-exchange-routing
python run_system.py

Next: Day 36 - Dead Letter Queues →

⭐ Found this valuable? Forward to a colleague learning distributed systems!

This lesson is part of the 254-Day Hands-On System Design series. View all lessons | Download source code | Join premium community

Hands On System Design - Distributed Systems Implementation

Day 35: Smart Log Routing with Exchange Types