| name | microservices-patterns |
| description | Design microservices architectures with service boundaries, event-driven communication, and resilience patterns. Use when building distributed systems, decomposing monoliths, or implementing microservices. |
Microservices Patterns
Master microservices architecture patterns including service boundaries, inter-service communication, data management, and resilience patterns for building distributed systems.
When to Use This Skill
- Decomposing monoliths into microservices
- Designing service boundaries and contracts
- Implementing inter-service communication
- Managing distributed data and transactions
- Building resilient distributed systems
- Implementing service discovery and load balancing
- Designing event-driven architectures
Core Concepts
1. Service Decomposition Strategies
By Business Capability
- Organize services around business functions
- Each service owns its domain
- Example: OrderService, PaymentService, InventoryService
By Subdomain (DDD)
- Core domain, supporting subdomains
- Bounded contexts map to services
- Clear ownership and responsibility
Strangler Fig Pattern
- Gradually extract from monolith
- New functionality as microservices
- Proxy routes to old/new systems
2. Communication Patterns
Synchronous (Request/Response)
Asynchronous (Events/Messages)
- Event streaming (Kafka)
- Message queues (RabbitMQ, SQS)
- Pub/Sub patterns
3. Data Management
Database Per Service
- Each service owns its data
- No shared databases
- Loose coupling
Saga Pattern
- Distributed transactions
- Compensating actions
- Eventual consistency
4. Resilience Patterns
Circuit Breaker
- Fail fast on repeated errors
- Prevent cascade failures
Retry with Backoff
- Transient fault handling
- Exponential backoff
Bulkhead
- Isolate resources
- Limit impact of failures
Service Decomposition Patterns
Pattern 1: By Business Capability
class OrderService:
"""Handles order lifecycle."""
async def create_order(self, order_data: dict) -> Order:
order = Order.create(order_data)
await self.event_bus.publish(
OrderCreatedEvent(
order_id=order.id,
customer_id=order.customer_id,
items=order.items,
total=order.total
)
)
return order
class PaymentService:
"""Handles payment processing."""
async def process_payment(self, payment_request: PaymentRequest) -> PaymentResult:
result = await self.payment_gateway.charge(
amount=payment_request.amount,
customer=payment_request.customer_id
)
if result.success:
await self.event_bus.publish(
PaymentCompletedEvent(
order_id=payment_request.order_id,
transaction_id=result.transaction_id
)
)
return result
class InventoryService:
"""Handles inventory management."""
async def reserve_items(self, order_id: , items: [OrderItem]) -> ReservationResult:
item items:
available = .inventory_repo.get_available(item.product_id)
available < item.quantity:
ReservationResult(
success=,
error=
)
reservation = .create_reservation(order_id, items)
.event_bus.publish(
InventoryReservedEvent(
order_id=order_id,
reservation_id=reservation.
)
)
ReservationResult(success=, reservation=reservation)
Pattern 2: API Gateway
from fastapi import FastAPI, HTTPException, Depends
import httpx
from circuitbreaker import circuit
app = FastAPI()
class APIGateway:
"""Central entry point for all client requests."""
def __init__(self):
self.order_service_url = "http://order-service:8000"
self.payment_service_url = "http://payment-service:8001"
self.inventory_service_url = "http://inventory-service:8002"
self.http_client = httpx.AsyncClient(timeout=5.0)
@circuit(failure_threshold=5, recovery_timeout=30)
async def call_order_service(self, path: str, method: str = "GET", **kwargs):
"""Call order service with circuit breaker."""
response = await self.http_client.request(
method,
f"{self.order_service_url}{path}",
**kwargs
)
response.raise_for_status()
return response.json()
async def create_order_aggregate(self, order_id: str) -> dict:
"""Aggregate data from multiple services."""
order, payment, inventory = asyncio.gather(
.call_order_service(),
.call_payment_service(),
.call_inventory_service(),
return_exceptions=
)
result = {: order}
(payment, Exception):
result[] = payment
(inventory, Exception):
result[] = inventory
result
():
:
order = gateway.call_order_service(
,
method=,
json=order_data
)
{: order}
httpx.HTTPError e:
HTTPException(status_code=, detail=)
Communication Patterns
Pattern 1: Synchronous REST Communication
import httpx
from tenacity import retry, stop_after_attempt, wait_exponential
class ServiceClient:
"""HTTP client with retries and timeout."""
def __init__(self, base_url: str):
self.base_url = base_url
self.client = httpx.AsyncClient(
timeout=httpx.Timeout(5.0, connect=2.0),
limits=httpx.Limits(max_keepalive_connections=20)
)
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10)
)
async def get(self, path: str, **kwargs):
"""GET with automatic retries."""
response = await self.client.get(f"{self.base_url}{path}", **kwargs)
response.raise_for_status()
return response.json()
async def post(self, path: str, **kwargs):
"""POST request."""
response = await self.client.post(f"", **kwargs)
response.raise_for_status()
response.json()
payment_client = ServiceClient()
result = payment_client.post(, json=payment_data)
Pattern 2: Asynchronous Event-Driven
from aiokafka import AIOKafkaProducer, AIOKafkaConsumer
import json
from dataclasses import dataclass, asdict
from datetime import datetime
@dataclass
class DomainEvent:
event_id: str
event_type: str
aggregate_id: str
occurred_at: datetime
data: dict
class EventBus:
"""Event publishing and subscription."""
def __init__(self, bootstrap_servers: List[str]):
self.bootstrap_servers = bootstrap_servers
self.producer = None
async def start(self):
self.producer = AIOKafkaProducer(
bootstrap_servers=self.bootstrap_servers,
value_serializer=lambda v: json.dumps(v).encode()
)
await self.producer.start()
async def publish(self, event: DomainEvent):
"""Publish event to Kafka topic."""
topic = event.event_type
await self.producer.send_and_wait(
topic,
value=asdict(event),
key=event.aggregate_id.encode()
)
async ():
consumer = AIOKafkaConsumer(
topic,
bootstrap_servers=.bootstrap_servers,
value_deserializer= v: json.loads(v.decode()),
group_id=
)
consumer.start()
:
message consumer:
event_data = message.value
handler(event_data)
:
consumer.stop()
():
order = save_order(order_data)
event = DomainEvent(
event_id=(uuid.uuid4()),
event_type=,
aggregate_id=order.,
occurred_at=datetime.now(),
data={
: order.,
: order.customer_id,
: order.total
}
)
event_bus.publish(event)
():
order_id = event_data[][]
items = event_data[][]
reserve_inventory(order_id, items)
Pattern 3: Saga Pattern (Distributed Transactions)
from enum import Enum
from typing import List, Callable
class SagaStep:
"""Single step in saga."""
def __init__(
self,
name: str,
action: Callable,
compensation: Callable
):
self.name = name
self.action = action
self.compensation = compensation
class SagaStatus(Enum):
PENDING = "pending"
COMPLETED = "completed"
COMPENSATING = "compensating"
FAILED = "failed"
class OrderFulfillmentSaga:
"""Orchestrated saga for order fulfillment."""
def __init__(self):
self.steps: List[SagaStep] = [
SagaStep(
"create_order",
action=self.create_order,
compensation=self.cancel_order
),
SagaStep(
"reserve_inventory",
action=self.reserve_inventory,
compensation=self.release_inventory
),
SagaStep(
"process_payment",
action=self.process_payment,
compensation=.refund_payment
),
SagaStep(
,
action=.confirm_order,
compensation=.cancel_order_confirmation
)
]
() -> SagaResult:
completed_steps = []
context = {: order_data}
:
step .steps:
result = step.action(context)
result.success:
.compensate(completed_steps, context)
SagaResult(
status=SagaStatus.FAILED,
error=result.error
)
completed_steps.append(step)
context.update(result.data)
SagaResult(status=SagaStatus.COMPLETED, data=context)
Exception e:
.compensate(completed_steps, context)
SagaResult(status=SagaStatus.FAILED, error=(e))
():
step (completed_steps):
:
step.compensation(context)
Exception e:
()
() -> StepResult:
order = order_service.create(context[])
StepResult(success=, data={: order.})
():
order_service.cancel(context[])
() -> StepResult:
result = inventory_service.reserve(
context[],
context[][]
)
StepResult(
success=result.success,
data={: result.reservation_id}
)
():
inventory_service.release(context[])
() -> StepResult:
result = payment_service.charge(
context[],
context[][]
)
StepResult(
success=result.success,
data={: result.transaction_id},
error=result.error
)
():
payment_service.refund(context[])
Resilience Patterns
Circuit Breaker Pattern
from enum import Enum
from datetime import datetime, timedelta
from typing import Callable, Any
class CircuitState(Enum):
CLOSED = "closed"
OPEN = "open"
HALF_OPEN = "half_open"
class CircuitBreaker:
"""Circuit breaker for service calls."""
def __init__(
self,
failure_threshold: int = 5,
recovery_timeout: int = 30,
success_threshold: int = 2
):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.success_threshold = success_threshold
self.failure_count = 0
self.success_count = 0
self.state = CircuitState.CLOSED
self.opened_at = None
async def call(self, func: Callable, *args, **kwargs) -> Any:
"""Execute function with circuit breaker."""
.state == CircuitState.OPEN:
._should_attempt_reset():
.state = CircuitState.HALF_OPEN
:
CircuitBreakerOpenError()
:
result = func(*args, **kwargs)
._on_success()
result
Exception e:
._on_failure()
():
.failure_count =
.state == CircuitState.HALF_OPEN:
.success_count +=
.success_count >= .success_threshold:
.state = CircuitState.CLOSED
.success_count =
():
.failure_count +=
.failure_count >= .failure_threshold:
.state = CircuitState.OPEN
.opened_at = datetime.now()
.state == CircuitState.HALF_OPEN:
.state = CircuitState.OPEN
.opened_at = datetime.now()
() -> :
(
datetime.now() - .opened_at
> timedelta(seconds=.recovery_timeout)
)
breaker = CircuitBreaker(failure_threshold=, recovery_timeout=)
():
breaker.call(
payment_client.process_payment,
payment_data
)
Resources
- references/service-decomposition-guide.md: Breaking down monoliths
- references/communication-patterns.md: Sync vs async patterns
- references/saga-implementation.md: Distributed transactions
- assets/circuit-breaker.py: Production circuit breaker
- assets/event-bus-template.py: Kafka event bus implementation
- assets/api-gateway-template.py: Complete API gateway
Best Practices
- Service Boundaries: Align with business capabilities
- Database Per Service: No shared databases
- API Contracts: Versioned, backward compatible
- Async When Possible: Events over direct calls
- Circuit Breakers: Fail fast on service failures
- Distributed Tracing: Track requests across services
- Service Registry: Dynamic service discovery
- Health Checks: Liveness and readiness probes
Common Pitfalls
- Distributed Monolith: Tightly coupled services
- Chatty Services: Too many inter-service calls
- Shared Databases: Tight coupling through data
- No Circuit Breakers: Cascade failures
- Synchronous Everything: Tight coupling, poor resilience
- Premature Microservices: Starting with microservices
- Ignoring Network Failures: Assuming reliable network
- No Compensation Logic: Can't undo failed transactions