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.
GitHub Link:
https://github.com/sysdr/course-p/tree/main/day35/day35-exchange-routingTraditional 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 messagesTopic 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,paymentRouting keys:
payment.processor.error,inventory.stock.updateTopic 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