How to Implement Event-Driven Architecture

Event-driven architecture decouples services by communicating through events instead of direct calls. It enables scalability, resilience, and real-time processing — but only if you handle the complexity correctly.

When to Use Event-Driven

✅ Good Fit

Scenario	Example
Decoupled microservices	Order placed → inventory, shipping, email all react independently
Real-time data processing	User action → analytics, recommendations update
Audit trail / compliance	Every state change persisted as an event
Async workloads	File uploaded → resize, scan, optimize in background

❌ Bad Fit

Scenario	Why Not	Better Approach
Simple CRUD	Over-engineering	Direct API call
Synchronous queries	User expects instant response	Request-response
Strong consistency required	Events are eventually consistent	Database transactions
Small team (< 5 devs)	Operational overhead	Monolith

Core Patterns

Event Sourcing

# Store events, derive state

class OrderEventStore:
    def __init__(self):
        self.events = []

    def append(self, event):
        event["timestamp"] = datetime.utcnow().isoformat()
        event["version"] = len(self.events) + 1
        self.events.append(event)

    def get_state(self, order_id):
        """Rebuild current state from events"""
        state = {"status": "unknown", "items": [], "total": 0}

        for event in self.events:
            if event["order_id"] != order_id:
                continue

            if event["type"] == "OrderCreated":
                state["status"] = "created"
                state["customer"] = event["customer_id"]
            elif event["type"] == "ItemAdded":
                state["items"].append(event["item"])
                state["total"] += event["price"]
            elif event["type"] == "OrderPaid":
                state["status"] = "paid"
            elif event["type"] == "OrderShipped":
                state["status"] = "shipped"

        return state

CQRS (Command Query Responsibility Segregation)

Commands (Write)              Events              Queries (Read)
┌──────────────┐     ┌────────────────┐     ┌──────────────┐
│ CreateOrder  │────▶│ OrderCreated   │────▶│ Order List   │
│ AddItem      │     │ ItemAdded      │     │ (Denormalized│
│ Pay          │     │ OrderPaid      │     │  read model) │
│ Ship         │     │ OrderShipped   │     │              │
└──────────────┘     └────────────────┘     └──────────────┘
  (Write DB)           (Event Store)          (Read DB)
  Normalized           Append-only            Optimized for
  for writes           immutable              queries

Step 1: Choose a Message Broker

Broker	Throughput	Ordering	Retention	Best For
Apache Kafka	Very High	Partition-level	Configurable (days-forever)	High-volume streaming
AWS SQS	High	FIFO available	14 days max	Simple queuing
RabbitMQ	Medium-High	Per-queue	Until consumed	Complex routing
AWS EventBridge	Medium	No guarantee	24 hours	AWS event routing
Redis Streams	Very High	Per-stream	Configurable	Low-latency

Step 2: Implement Idempotent Consumers

# Every event consumer MUST be idempotent
# (processing the same event twice should be safe)

class IdempotentConsumer:
    def __init__(self, db):
        self.db = db

    def process(self, event):
        event_id = event["id"]

        # Check if already processed
        if self.db.execute(
            "SELECT 1 FROM processed_events WHERE event_id = %s",
            (event_id,)
        ).fetchone():
            print(f"Event {event_id} already processed, skipping")
            return

        # Process the event
        self._handle(event)

        # Mark as processed
        self.db.execute(
            "INSERT INTO processed_events (event_id, processed_at) VALUES (%s, NOW())",
            (event_id,)
        )
        self.db.commit()

    def _handle(self, event):
        if event["type"] == "OrderPaid":
            # Send confirmation email, update inventory, etc.
            pass

Step 3: Saga Pattern for Distributed Transactions

# Orchestration saga for order processing
class OrderSaga:
    steps = [
        {"action": "reserve_inventory", "compensate": "release_inventory"},
        {"action": "charge_payment", "compensate": "refund_payment"},
        {"action": "create_shipment", "compensate": "cancel_shipment"},
        {"action": "send_confirmation", "compensate": "send_cancellation"},
    ]

    async def execute(self, order):
        completed = []
        for step in self.steps:
            try:
                await getattr(self, step["action"])(order)
                completed.append(step)
            except Exception as e:
                # Compensate in reverse order
                for comp_step in reversed(completed):
                    await getattr(self, comp_step["compensate"])(order)
                raise SagaFailed(f"Step {step['action']} failed: {e}")

Common Pitfalls

Pitfall	Problem	Solution
No idempotency	Duplicate processing	Deduplication by event ID
No dead letter queue	Lost messages	Configure DLQ on every queue
No schema versioning	Breaking consumers	Schema registry (Avro/Protobuf)
Eventual consistency surprise	Users see stale data	UI shows “processing” state
Event ordering assumptions	Out-of-order processing	Use partition keys, handle reordering

Event-Driven Architecture Checklist

:::note[Source] This guide is derived from operational intelligence at Garnet Grid Consulting. For architecture consulting, visit garnetgrid.com. :::