Microservices Expert
Expert guidance for microservices architecture, design patterns, service communication, and distributed system challenges.
Core Concepts
Microservices Principles
- Single responsibility per service
- Independently deployable
- Decentralized data management
- Infrastructure automation
- Design for failure
- Evolutionary design
Architecture Patterns
- API Gateway
- Service Discovery
- Circuit Breaker
- Saga Pattern
- Event Sourcing
- CQRS
Communication
- Synchronous (HTTP/REST, gRPC)
- Asynchronous (Message queues, Events)
- Service mesh
- API composition
- Backend for Frontend (BFF)
Service Design
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import httpx
from typing import List, Optional
from circuitbreaker import circuit
import asyncio
# Individual Microservice
app = FastAPI(title="Order Service", version="1.0.0")
class Order(BaseModel):
id: str
user_id: str
items: List[dict]
total: float
status: str
class OrderService:
def __init__(self, inventory_url: str, payment_url: str):
self.inventory_url = inventory_url
self.payment_url = payment_url
self.client = httpx.AsyncClient()
@circuit(failure_threshold=5, recovery_timeout=60)
async def check_inventory(self, items: List[dict]) -> bool:
"""Check inventory availability with circuit breaker"""
try:
response = await self.client.post(
f"{self.inventory_url}/check",
json={"items": items},
timeout=5.0
)
return response.json()["available"]
except Exception as e:
print(f"Inventory service error: {e}")
raise
@circuit(failure_threshold=5, recovery_timeout=60)
async def process_payment(self, user_id: str, amount: float) -> dict:
"""Process payment with circuit breaker"""
try:
response = await self.client.post(
f"{self.payment_url}/charge",
json={"user_id": user_id, "amount": amount},
timeout=10.0
)
return response.json()
except Exception as e:
print(f"Payment service error: {e}")
raise
async def create_order(self, order: Order) -> Order:
"""Create order with coordination"""
# 1. Check inventory
inventory_available = await self.check_inventory(order.items)
if not inventory_available:
raise HTTPException(400, "Items not available")
# 2. Process payment
payment = await self.process_payment(order.user_id, order.total)
if payment["status"] != "success":
raise HTTPException(400, "Payment failed")
# 3. Reserve inventory
await self.reserve_inventory(order.items)
# 4. Create order record
order.status = "confirmed"
await self.save_order(order)
return order
@app.post("/orders", response_model=Order)
async def create_order(order: Order):
service = OrderService(
inventory_url="http://inventory-service",
payment_url="http://payment-service"
)
return await service.create_order(order)Saga Pattern (Distributed Transactions)
from enum import Enum
from typing import List, Callable
import asyncio
class SagaStep:
def __init__(self, action: Callable, compensation: Callable):
self.action = action
self.compensation = compensation
class SagaOrchestrator:
"""Orchestrate distributed transactions using Saga pattern"""
def __init__(self):
self.steps: List[SagaStep] = []
self.completed_steps: List[SagaStep] = []
def add_step(self, action: Callable, compensation: Callable):
"""Add a step to the saga"""
self.steps.append(SagaStep(action, compensation))
async def execute(self) -> bool:
"""Execute saga with compensation on failure"""
try:
# Execute all steps
for step in self.steps:
result = await step.action()
self.completed_steps.append(step)
if not result:
await self.compensate()
return False
return True
except Exception as e:
print(f"Saga failed: {e}")
await self.compensate()
return False
async def compensate(self):
"""Rollback completed steps"""
# Execute compensations in reverse order
for step in reversed(self.completed_steps):
try:
await step.compensation()
except Exception as e:
print(f"Compensation failed: {e}")
# Example: Order Saga
class OrderSaga:
async def create_order_with_saga(self, order_data: dict):
saga = SagaOrchestrator()
# Step 1: Reserve inventory
saga.add_step(
action=lambda: self.reserve_inventory(order_data["items"]),
compensation=lambda: self.release_inventory(order_data["items"])
)
# Step 2: Process payment
saga.add_step(
action=lambda: self.charge_payment(order_data["user_id"], order_data["total"]),
compensation=lambda: self.refund_payment(order_data["user_id"], order_data["total"])
)
# Step 3: Create order
saga.add_step(
action=lambda: self.create_order_record(order_data),
compensation=lambda: self.delete_order_record(order_data["id"])
)
# Execute saga
success = await saga.execute()
if success:
await self.send_confirmation(order_data["user_id"])
return {"status": "success", "order_id": order_data["id"]}
else:
return {"status": "failed", "message": "Order creation failed"}Service Discovery
import consul
from typing import List, Optional
import random
class ServiceRegistry:
"""Service discovery using Consul"""
def __init__(self, consul_host: str = "localhost", consul_port: int = 8500):
self.consul = consul.Consul(host=consul_host, port=consul_port)
def register_service(self, service_name: str, service_id: str,
host: str, port: int, tags: List[str] = None):
"""Register service with Consul"""
self.consul.agent.service.register(
name=service_name,
service_id=service_id,
address=host,
port=port,
tags=tags or [],
check=consul.Check.http(
f"http://{host}:{port}/health",
interval="10s",
timeout="5s"
)
)
def deregister_service(self, service_id: str):
"""Deregister service"""
self.consul.agent.service.deregister(service_id)
def discover_service(self, service_name: str) -> Optional[dict]:
"""Discover healthy service instance"""
_, services = self.consul.health.service(service_name, passing=True)
if not services:
return None
# Load balance: random selection
service = random.choice(services)
return {
"id": service["Service"]["ID"],
"address": service["Service"]["Address"],
"port": service["Service"]["Port"],
"tags": service["Service"]["Tags"]
}
def get_all_services(self, service_name: str) -> List[dict]:
"""Get all healthy instances of a service"""
_, services = self.consul.health.service(service_name, passing=True)
return [
{
"id": s["Service"]["ID"],
"address": s["Service"]["Address"],
"port": s["Service"]["Port"]
}
for s in services
]API Gateway
from fastapi import FastAPI, Request, Response
import httpx
from typing import Dict
import jwt
app = FastAPI(title="API Gateway")
class APIGateway:
"""API Gateway for routing and cross-cutting concerns"""
def __init__(self):
self.service_registry = ServiceRegistry()
self.client = httpx.AsyncClient()
async def route_request(self, service: str, path: str,
method: str, **kwargs) -> Response:
"""Route request to appropriate microservice"""
# Discover service
service_info = self.service_registry.discover_service(service)
if not service_info:
return Response(
content={"error": "Service unavailable"},
status_code=503
)
# Build URL
url = f"http://{service_info['address']}:{service_info['port']}{path}"
# Forward request
response = await self.client.request(method, url, **kwargs)
return Response(
content=response.content,
status_code=response.status_code,
headers=dict(response.headers)
)
async def authenticate(self, token: str) -> Optional[dict]:
"""Centralized authentication"""
try:
payload = jwt.decode(token, "secret", algorithms=["HS256"])
return payload
except jwt.JWTError:
return None
async def rate_limit(self, client_id: str) -> bool:
"""Centralized rate limiting"""
# Implementation using Redis
pass
# Gateway endpoints
@app.api_route("/{service}/{path:path}", methods=["GET", "POST", "PUT", "DELETE"])
async def gateway_route(service: str, path: str, request: Request):
gateway = APIGateway()
# Authentication
token = request.headers.get("Authorization", "").replace("Bearer ", "")
user = await gateway.authenticate(token)
if not user:
return Response(content={"error": "Unauthorized"}, status_code=401)
# Rate limiting
if not await gateway.rate_limit(user["id"]):
return Response(content={"error": "Rate limit exceeded"}, status_code=429)
# Route request
return await gateway.route_request(
service=service,
path=f"/{path}",
method=request.method,
headers=dict(request.headers),
content=await request.body()
)Event-Driven Architecture
import pika
import json
from typing import Callable, Dict
import asyncio
class EventBus:
"""Message broker for event-driven communication"""
def __init__(self, rabbitmq_url: str):
self.connection = pika.BlockingConnection(
pika.URLParameters(rabbitmq_url)
)
self.channel = self.connection.channel()
self.handlers: Dict[str, Callable] = {}
def publish_event(self, event_type: str, data: dict):
"""Publish event to all subscribers"""
self.channel.exchange_declare(
exchange='events',
exchange_type='topic',
durable=True
)
message = json.dumps({
"event_type": event_type,
"data": data,
"timestamp": datetime.now().isoformat()
})
self.channel.basic_publish(
exchange='events',
routing_key=event_type,
body=message,
properties=pika.BasicProperties(
delivery_mode=2 # persistent
)
)
def subscribe(self, event_type: str, handler: Callable):
"""Subscribe to event type"""
self.handlers[event_type] = handler
# Declare queue
queue_name = f"{event_type}_queue"
self.channel.queue_declare(queue=queue_name, durable=True)
# Bind queue to exchange
self.channel.queue_bind(
queue=queue_name,
exchange='events',
routing_key=event_type
)
# Start consuming
def callback(ch, method, properties, body):
message = json.loads(body)
handler(message["data"])
ch.basic_ack(delivery_tag=method.delivery_tag)
self.channel.basic_consume(
queue=queue_name,
on_message_callback=callback
)
def start_consuming(self):
"""Start consuming events"""
self.channel.start_consuming()
# Usage Example
event_bus = EventBus("amqp://localhost")
# Service A publishes event
event_bus.publish_event("order.created", {
"order_id": "12345",
"user_id": "user_1",
"total": 99.99
})
# Service B subscribes to event
def handle_order_created(data):
print(f"Processing order: {data['order_id']}")
# Send email, update inventory, etc.
event_bus.subscribe("order.created", handle_order_created)Best Practices
Design
- Keep services small and focused
- Design for failure (circuit breakers, retries)
- Use asynchronous communication when possible
- Implement proper service boundaries
- Avoid distributed monoliths
- Use API versioning
- Implement health checks
Data Management
- Database per service
- Use eventual consistency
- Implement saga pattern for distributed transactions
- Use event sourcing for audit trails
- Cache aggressively
- Avoid distributed joins
Operations
- Implement distributed tracing
- Centralized logging
- Monitor service health
- Automate deployments
- Use service mesh for cross-cutting concerns
- Implement feature flags
- Practice chaos engineering
Anti-Patterns
❌ Distributed monolith ❌ Shared database between services ❌ Synchronous communication everywhere ❌ No service versioning ❌ Tight coupling between services ❌ No circuit breakers ❌ Missing distributed tracing
Resources
- Microservices Patterns: https://microservices.io/patterns/
- Building Microservices (book)
- Service Mesh: https://istio.io/
- Consul: https://www.consul.io/
- RabbitMQ: https://www.rabbitmq.com/