Mini Project: Multi-Service Event Bus
In this tutorial, you will learn about Mini Project: Multi. We cover key concepts, practical examples, and best practices to help you master this topic.
Build a multi-service event bus using RabbitMQ topic exchanges that decouples Microservices with reliable message delivery, routing, and monitoring.
What You'll Learn
By the end of this project, you will build a complete event bus system with multiple producers and consumers, topic-based routing, persistent messages, dead letter handling, and a monitoring dashboard.
Why It Matters
A well-designed event bus is the backbone of a microservice architecture. It decouples services, enables asynchronous communication, and provides a single point for monitoring and reliability guarantees. This project brings together all the patterns you learned.
Real-World Use
Doda Browser's backend uses an event bus to coordinate file upload, malware scanning, thumbnail generation, and notification delivery. Each service publishes and consumes events through a central topic exchange, enabling independent scaling and deployment.
Architecture
flowchart TB
subgraph "Services"
U[Upload Service]
S[Scanner Service]
T[Thumbnail Service]
N[Notification Service]
end
subgraph "Event Bus"
EX[Topic Exchange: dodatech.events]
Q1[scan.requests]
Q2[scan.results]
Q3[thumbnail.requests]
Q4[notifications]
end
subgraph "Dead Letter"
DLQ[Dead Letter Queue]
end
U -->|"file.uploaded"| EX
EX --> Q1 --> S
S -->|"scan.completed"| EX
EX --> Q3 --> T
EX --> Q4 --> N
S -->|"scan.failed"| DLQ
style EX fill:#f90,color:#fff
style DLQ fill:#e74c3c,color:#fff
Step 1: Event Bus Setup
# event_bus.py
import pika
import json
import uuid
class EventBus:
def __init__(self, url='amqp://localhost'):
self.url = url
self.exchange = 'dodatech.events'
self.connection = None
self.channel = None
def connect(self):
self.connection = pika.BlockingConnection(pika.ConnectionParameters(self.url))
self.channel = self.connection.channel()
self.channel.exchange_declare(
exchange=self.exchange,
exchange_type='topic',
durable=True
)
print(f"[EventBus] Connected, exchange: {self.exchange}")
return self
def publish(self, routing_key, data):
message = {
'event_id': str(uuid.uuid4()),
'timestamp': __import__('time').time(),
'data': data,
}
self.channel.basic_publish(
exchange=self.exchange,
routing_key=routing_key,
body=json.dumps(message),
properties=pika.BasicProperties(
delivery_mode=2,
content_type='application/json',
message_id=message['event_id'],
)
)
print(f"[EventBus] Published: {routing_key} ({message['event_id'][:8]})")
return message['event_id']
def subscribe(self, queue_name, binding_keys, handler):
self.channel.queue_declare(queue=queue_name, durable=True)
for key in binding_keys:
self.channel.queue_bind(
exchange=self.exchange,
queue=queue_name,
routing_key=key
)
print(f"[EventBus] Bound {queue_name} <- {key}")
def callback(ch, method, properties, body):
event = json.loads(body)
handler(event, method)
ch.basic_ack(delivery_tag=method.delivery_tag)
self.channel.basic_consume(queue=queue_name, on_message_callback=callback)
return queue_name
def start_consuming(self):
self.channel.start_consuming()
def close(self):
if self.channel:
self.channel.close()
if self.connection:
self.connection.close()
Step 2: Services
# services.py
import time
import random
from event_bus import EventBus
import threading
bus = EventBus()
bus.connect()
def upload_service():
files = ['photo.jpg', 'document.pdf', 'video.mp4']
for f in files:
bus.publish('file.uploaded', {
'file': f,
'size': random.randint(100000, 10000000),
'user_id': random.randint(1, 100),
})
time.sleep(0.5)
print("[Upload] Published 3 file uploads")
def scanner_service():
def handle(event, method):
data = event['data']
threat = random.random() < 0.2
print(f"[Scanner] Scanning: {data['file']}")
time.sleep(random.uniform(0.5, 1.5))
if threat:
bus.publish('scan.failed', {
'file': data['file'],
'reason': 'Malware detected',
'user_id': data['user_id'],
})
else:
bus.publish('scan.completed', {
'file': data['file'],
'result': 'clean',
'user_id': data['user_id'],
})
print(f"[Scanner] Done: {data['file']}")
bus.subscribe('scan_requests', ['file.uploaded'], handle)
bus.start_consuming()
def notification_service():
def handle(event, method):
data = event['data']
key = method.routing_key
print(f"[Notifier] {'Threat detected!' if 'failed' in key else 'File scanned'}: {data['file']}")
bus.subscribe('notifications', ['scan.completed', 'scan.failed'], handle)
bus.start_consuming()
threads = [
threading.Thread(target=scanner_service, daemon=True),
threading.Thread(target=notification_service, daemon=True),
]
for t in threads:
t.start()
time.sleep(1)
upload_service()
time.sleep(3)
bus.close()
Expected output:
[EventBus] Connected, exchange: dodatech.events
[EventBus] Bound scan_requests <- file.uploaded
[EventBus] Bound notifications <- scan.completed
[EventBus] Bound notifications <- scan.failed
[EventBus] Published: file.uploaded (a1b2c3d4)
[Scanner] Scanning: photo.jpg
[Scanner] Done: photo.jpg
[EventBus] Published: scan.completed (e5f6g7h8)
[Notifier] File scanned: photo.jpg
...
Step 3: Dead Letter and Monitoring
# monitoring.py
import pika
import json
class Monitor:
def __init__(self):
self.conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
self.ch = self.conn.channel()
def queue_depth(self, queue_name):
q = self.ch.queue_declare(queue=queue_name, passive=True)
return q.method.message_count
def report(self):
queues = ['scan_requests', 'notifications', 'dead_letter_queue']
print("\n=== Queue Status ===")
for q in queues:
try:
depth = self.queue_depth(q)
print(f" {q}: {depth} messages")
except:
print(f" {q}: not found")
mon = Monitor()
mon.report()
Expected output:
=== Queue Status ===
scan_requests: 2 messages
notifications: 0 messages
dead_letter_queue: 0 messages
Challenge
Extend the event bus to handle: (1) exactly-once delivery with idempotent consumers, (2) message replay from a log, (3) priority queuing for scan requests from premium users, (4) a dashboard showing real-time event flow, and (5) automated retry with exponential backoff for failed scans.
FAQ
What's Next
You have completed the Message Queue Patterns series. Next, explore RabbitMQ for a deep dive into the most popular Message Broker, or jump to Celery for Python-based task queue implementation.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro