Event-Driven Architecture

What is Event-Driven Architecture?

A design paradigm where services communicate by producing and consuming events asynchronously.

In traditional request/response systems, Service A directly calls Service B and waits for a response. This creates tight coupling and synchronous dependencies. In contrast, Event-Driven Architecture (EDA) uses events as the primary means of communication between services.

An event is a record of something that happened in the system (e.g., "OrderPlaced", "PaymentProcessed", "UserRegistered"). Services publish events to a message broker, and other services subscribe to events they care about—without knowing who produced them.

Event Patterns

Different patterns suit different use cases. Here are the most common event-driven patterns.

Kafka vs RabbitMQ: Detailed Comparison

Both are popular message brokers, but they're optimized for different use cases. Understanding their trade-offs is critical for system design.

Delivery Guarantees

Understanding delivery semantics is crucial for building reliable event-driven systems. The choice affects performance, complexity, and data consistency.

Idempotency Patterns

Since at-least-once delivery causes duplicates, consumers must handle the same event multiple times safely. This is called idempotency.

Why Idempotency Matters

Imagine an "OrderPlaced" event causes a charge to the customer's credit card. If the event is processed twice, the customer gets charged twice—catastrophic! Idempotency ensures repeated processing is safe.

Deduplication Strategies

• Unique Message ID: Each event has a unique ID. Store processed IDs in a database/cache (Redis). Reject duplicates.
• Natural Idempotency: Some operations are naturally idempotent (e.g., "SET user_status = ACTIVE").
• Versioning: Use event timestamps or version numbers. Only process events newer than the last processed version.
• Database Constraints: Use UNIQUE constraints to prevent duplicate inserts.

Python Example: Idempotent Message Handler

import redis
import json

# Redis for tracking processed message IDs
redis_client = redis.StrictRedis(host='localhost', port=6379, decode_responses=True)

def process_event(event):
    """
    Idempotent event handler using Redis for deduplication.
    """
    event_id = event['id']
    event_type = event['type']
    
    # Check if we've already processed this event
    cache_key = f"processed:{event_id}"
    if redis_client.exists(cache_key):
        print(f"⚠️  Duplicate event {event_id} detected. Skipping.")
        return
    
    # Process the event (business logic)
    if event_type == "OrderPlaced":
        print(f"✅ Processing order: {event['order_id']}")
        # Charge credit card, update inventory, etc.
    
    # Mark event as processed (TTL: 7 days)
    redis_client.setex(cache_key, 604800, "processed")
    print(f"✅ Event {event_id} processed successfully.")

# Simulate consuming events
events = [
    {"id": "evt_001", "type": "OrderPlaced", "order_id": "ORD123"},
    {"id": "evt_001", "type": "OrderPlaced", "order_id": "ORD123"},  # Duplicate
    {"id": "evt_002", "type": "OrderPlaced", "order_id": "ORD456"}
]

for event in events:
    process_event(event)

# Output:
# ✅ Processing order: ORD123
# ✅ Event evt_001 processed successfully.
# ⚠️  Duplicate event evt_001 detected. Skipping.
# ✅ Processing order: ORD456
# ✅ Event evt_002 processed successfully.

Error Handling & Reliability

Events can fail to process due to bugs, transient errors, or downstream service failures. A robust event-driven system handles these gracefully.

Schema Registry & Event Contracts

As systems evolve, event schemas change (new fields, renamed fields, etc.). A Schema Registry ensures producers and consumers agree on the event format and handles versioning.

Why Schema Registry?

• Type Safety: Prevents runtime errors from schema mismatches.
• Evolution: Add fields without breaking existing consumers (forward/backward compatibility).
• Documentation: Schema serves as living documentation of event structure.

Popular Formats

Python Implementation Examples

Kafka Producer & Consumer

Using kafka-python library to produce and consume events.

from kafka import KafkaProducer, KafkaConsumer
import json

# === PRODUCER ===
producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

# Publish an event
event = {
    "event_id": "evt_12345",
    "type": "OrderPlaced",
    "order_id": "ORD789",
    "user_id": "user_456",
    "amount": 99.99
}

producer.send('orders', value=event, key=b'user_456')
producer.flush()
print("✅ Event published to Kafka")

# === CONSUMER ===
consumer = KafkaConsumer(
    'orders',
    bootstrap_servers=['localhost:9092'],
    auto_offset_reset='earliest',
    enable_auto_commit=True,
    group_id='order-processor',
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

print("🔄 Consuming events...")
for message in consumer:
    event = message.value
    print(f"📩 Received: {event['type']} | Order: {event['order_id']}")
    # Process event (idempotent handler recommended)

RabbitMQ Producer & Consumer

Using pika library for traditional message queuing.

import pika
import json

# === PRODUCER ===
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# Declare a queue
channel.queue_declare(queue='notifications', durable=True)

message = {
    "user_id": "user_789",
    "type": "EmailNotification",
    "subject": "Your order was shipped!",
    "body": "Track your package at ..."
}

channel.basic_publish(
    exchange='',
    routing_key='notifications',
    body=json.dumps(message),
    properties=pika.BasicProperties(delivery_mode=2)  # Persistent
)

print("✅ Message sent to RabbitMQ")
connection.close()

# === CONSUMER ===
def callback(ch, method, properties, body):
    message = json.loads(body)
    print(f"📧 Sending {message['type']} to user {message['user_id']}")
    # Send email via SMTP, etc.
    ch.basic_ack(delivery_tag=method.delivery_tag)

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='notifications', durable=True)

channel.basic_consume(queue='notifications', on_message_callback=callback)
print("🔄 Waiting for messages...")
channel.start_consuming()