Microservices with Kafka And RabbitMQ

Microservices Architecture with Message Brokers

The Extended Stack Overview

In this architecture:

• FastAPI serves as the HTTP gateway and individual microservice framework

• PostgreSQL provides persistent data storage for each service's domain data

• Redis handles caching, sessions, and real-time features

• Kafka/RabbitMQ enables asynchronous communication between services

Message Brokers: Kafka vs RabbitMQ

RabbitMQ: The General-Purpose Message Broker

When to Choose RabbitMQ:

• Complex routing needs

• Background tasks/job queues

• Request/reply patterns

• When you need message acknowledgments and delivery guarantees

Key Features:

• Exchanges & Queues: Direct, topic, fanout, headers exchanges

• Protocols: AMQP 0-9-1 primarily

• Delivery Guarantees: At-least-once, exactly-once(with publisher confirms)

• Message Priorities: Yes

• Dead Letter Exchanges: For handling failed messages

Common Microservices Patterns with RabbitMQ:

python

# Example: Order Service publishing an event
async def create_order(order_data):
    # Create order in local PostgreSQL
    order = await OrderService.create(order_data)
    
    # Publish event to RabbitMQ
    await rabbitmq_publish(
        exchange="order_events",
        routing_key="order.created",
        message={
            "event_id": str(uuid.uuid4()),
            "event_type": "order_created",
            "timestamp": datetime.utcnow().isoformat(),
            "data": {
                "order_id": order.id,
                "user_id": order.user_id,
                "total_amount": order.total_amount
            }
        }
    )
    
    return order

# Inventory Service consuming the event
async def consume_order_created():
    async def callback(message):
        event = json.loads(message.body)
        if event["event_type"] == "order_created":
            await InventoryService.reserve_stock(
                event["data"]["order_id"],
                event["data"]["items"]
            )
    
    await rabbitmq_consume(
        queue="inventory_order_created",
        exchange="order_events",
        routing_key="order.created",
        callback=callback
    )

Apache Kafka: The Distributed Event Streaming Platform

When to Choose Kafka:

• High throughput event streaming

• Event sourcing architectures

• Real-time data processing pipelines

• When you need replayability and long-term retention

• Multiple consumers needing the same messages

Key Features:

• Topics & Partitions: Scalable, ordered message streams

• Consumer Groups: Parallel message processing

• Retention Period: Configurable (hours to forever)

• High Throughput: Millions of messages per second

• Durability: Replicated, persistent log

Common Microservices Patterns with Kafka:

python

# Example: User Service publishing to Kafka
async def update_user_profile(user_id, profile_data):
    # Update in local PostgreSQL
    user = await UserService.update(user_id, profile_data)
    
    # Produce event to Kafka
    await kafka_producer.send(
        topic="user-events",
        value={
            "event_id": str(uuid.uuid4()),
            "event_type": "user_profile_updated",
            "timestamp": datetime.utcnow().isoformat(),
            "entity_type": "user",
            "entity_id": user_id,
            "data": {
                "email": user.email,
                "name": user.name,
                "preferences": user.preferences
            }
        }
    )

# Multiple services consuming the same event
# Notification Service
async def send_profile_update_notification():
    async for message in kafka_consumer:
        event = message.value
        if event["event_type"] == "user_profile_updated":
            await NotificationService.send(
                event["entity_id"],
                "Profile updated successfully"
            )

# Analytics Service
async def track_profile_updates():
    async for message in kafka_consumer:
        event = message.value
        if event["event_type"] == "user_profile_updated":
            await AnalyticsService.track(
                "profile_update",
                event["entity_id"],
                event["data"]
            )

Complete Microservices Architecture Example

Service Structure

text

ecommerce-platform/
├── api-gateway/                 # FastAPI
├── services/
│   ├── user-service/           # FastAPI + PostgreSQL
│   ├── order-service/          # FastAPI + PostgreSQL  
│   ├── inventory-service/      # FastAPI + PostgreSQL
│   ├── notification-service/   # FastAPI + Redis
│   └── analytics-service/      # FastAPI + ClickHouse
├── shared/
│   ├── events/                 # Shared event schemas
│   └── clients/               # Service clients
└── infrastructure/
    ├── kafka/                 # Kafka cluster
    ├── redis/                 # Redis cluster
    └── postgres/             # PostgreSQL instances

Event-Driven Order Processing Flow

python

# Order Service
async def create_order(order_data):
    # 1. Validate business rules
    if not await InventoryService.check_availability(order_data.items):
        raise HTTPException(400, "Items not available")
    
    # 2. Create order in local database
    async with database.transaction():
        order = await Order.create(**order_data)
        order_items = await OrderItem.bulk_create(order.id, order_data.items)
    
    # 3. Publish OrderCreated event
    await kafka_producer.send(
        topic="order-events",
        value={
            "event_id": str(uuid.uuid4()),
            "event_type": "OrderCreated",
            "timestamp": datetime.utcnow().isoformat(),
            "data": {
                "order_id": order.id,
                "user_id": order.user_id,
                "items": order_data.items,
                "total_amount": order.total_amount
            }
        }
    )
    
    # 4. Cache order summary in Redis
    await redis.setex(
        f"order:summary:{order.id}",
        3600,  # 1 hour
        json.dumps(order.summary())
    )
    
    return order

# Inventory Service (consumes OrderCreated)
async def handle_order_created():
    consumer = aiokafka.AIOKafkaConsumer(
        "order-events",
        bootstrap_servers=KAFKA_BROKERS,
        group_id="inventory-service"
    )
    await consumer.start()
    
    try:
        async for msg in consumer:
            event = json.loads(msg.value)
            if event["event_type"] == "OrderCreated":
                await reserve_stock(event["data"])
                
                # Publish StockReserved event
                await kafka_producer.send(
                    topic="inventory-events",
                    value={
                        "event_type": "StockReserved",
                        "order_id": event["data"]["order_id"]
                    }
                )
    finally:
        await consumer.stop()

# Payment Service (consumes StockReserved from RabbitMQ)
async def process_payment():
    async with rabbitmq_channel() as channel:
        await channel.queue_declare("payment-queue")
        
        async for message in channel:
            event = json.loads(message.body)
            if event["event_type"] == "StockReserved":
                payment_result = await PaymentService.charge(
                    event["order_id"],
                    event["amount"]
                )
                
                if payment_result.success:
                    # Publish PaymentCompleted event
                    await kafka_producer.send(
                        topic="payment-events",
                        value={
                            "event_type": "PaymentCompleted",
                            "order_id": event["order_id"]
                        }
                    )
                else:
                    # Compensating transaction - release stock
                    await kafka_producer.send(
                        topic="inventory-events", 
                        value={
                            "event_type": "StockRelease",
                            "order_id": event["order_id"]
                        }
                    )

Technology Stack & Libraries

FastAPI Microservices Setup

python

# requirements.txt for a typical service
fastapi==0.104.1
uvicorn==0.24.0
# Database
asyncpg==0.29.0
sqlalchemy==2.0.23
alembic==1.12.1
# Message Brokers
aiokafka==0.8.0
aio-pika==9.4.1
# Redis
redis==5.0.1
# HTTP clients
httpx==0.25.2

Configuration Management

python

# app/core/config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    # Service config
    service_name: str = "user-service"
    environment: str = "development"
    
    # Database
    database_url: str
    
    # Redis
    redis_url: str
    
    # Kafka
    kafka_brokers: list[str] = ["localhost:9092"]
    
    # RabbitMQ
    rabbitmq_url: str = "amqp://guest:guest@localhost/"
    
    class Config:
        env_file = ".env"

Service Discovery & Communication

python

# HTTP-based service communication
async def call_service(service_name: str, endpoint: str, method: str = "GET", data: dict = None):
    # Service discovery via Consul or Kubernetes DNS
    service_url = await service_discovery.resolve(service_name)
    
    async with httpx.AsyncClient() as client:
        response = await client.request(
            method,
            f"{service_url}/{endpoint}",
            json=data,
            headers={"X-Service-Origin": settings.service_name}
        )
        return response.json()

# Event-based communication
async def publish_event(topic: str, event: dict):
    await kafka_producer.send(topic, value=event)

Deployment & Orchestration

Docker Compose for Development

yaml

# docker-compose.yml
version: '3.8'
services:
  # Message Brokers
  zookeeper:
    image: confluentinc/cp-zookeeper:latest
    # ... config
  
  kafka:
    image: confluentinc/cp-kafka:latest
    depends_on: [zookeeper]
    # ... config
  
  rabbitmq:
    image: rabbitmq:3-management
    # ... config
  
  # Databases
  postgres-user:
    image: postgres:15
    # ... config
  
  postgres-order:
    image: postgres:15
    # ... config
  
  redis:
    image: redis:7-alpine
  
  # Services
  api-gateway:
    build: ./api-gateway
    ports: ["8000:8000"]
    depends_on: [kafka, rabbitmq]
  
  user-service:
    build: ./services/user-service
    depends_on: [postgres-user, redis, kafka]
  
  order-service:
    build: ./services/order-service  
    depends_on: [postgres-order, redis, kafka, rabbitmq]

Kubernetes for Production

yaml

# k8s/service.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: order-service
spec:
  replicas: 3
  selector:
    matchLabels:
      app: order-service
  template:
    spec:
      containers:
      - name: order-service
        image: order-service:latest
        env:
        - name: KAFKA_BROKERS
          value: "kafka-cluster:9092"
        - name: DATABASE_URL
          valueFrom:
            secretKeyRef:
              name: order-db-secret
              key: url
---
apiVersion: v1
kind: Service
metadata:
  name: order-service
spec:
  selector:
    app: order-service
  ports:
  - port: 80
    targetPort: 8000

Key Benefits of This Architecture

1. Loose Coupling

• Services communicate via events, not direct calls

• Each service can evolve independently

• Technology heterogeneity per service

2. Scalability

• Scale services independently based on load

• Kafka enables high-throughput event processing

• Redis provides fast data access

3. Resilience

• Message brokers provide retry mechanisms

• Failed messages can be re-processed

• Circuit breakers prevent cascading failures

4. Data Consistency

• Saga pattern for distributed transactions

• Event sourcing for auditability

• Compensating transactions for rollbacks

5. Observability

• Distributed tracing across services

• Centralized logging

• Metrics collection and monitoring

When to Use Which Broker?

Use RabbitMQ When:

• You need complex routing and message patterns

• Task queues and background job processing

• Request-reply communication patterns

• Smaller to medium scale systems

Use Kafka When:

• High-throughput event streaming

• Multiple consumers need the same messages

• Event sourcing and replayability

• Real-time analytics and data pipelines

• Large-scale, data-intensive applications

Use Both When:

• Kafka for event streaming and data pipeline

• RabbitMQ for task queues and inter-service RPC

• Leverage strengths of both systems

This extended stack provides a robust foundation for building modern, scalable, and maintainable microservices architectures that can handle complex business domains and high traffic loads.