Modèles de Microservices
Maîtrisez les modèles d'architecture microservices incluant les limites de service, la communication inter-services, la gestion des données et les modèles de résilience pour construire des systèmes distribués.
Quand Utiliser Cette Compétence
- Décomposer les monolithes en microservices
- Concevoir les limites et contrats de service
- Mettre en œuvre la communication inter-services
- Gérer les données distribuées et les transactions
- Construire des systèmes distribués résilients
- Mettre en œuvre la découverte de service et l'équilibrage de charge
- Concevoir des architectures orientées événements
Concepts Fondamentaux
1. Stratégies de Décomposition de Service
Par Capacité Métier
- Organiser les services autour des fonctions métier
- Chaque service possède son domaine
- Exemple : OrderService, PaymentService, InventoryService
Par Sous-domaine (DDD)
- Domaine cœur, sous-domaines support
- Les contextes limités correspondent aux services
- Responsabilité et propriété claires
Modèle Strangler Fig
- Extraire progressivement du monolithe
- Nouvelle fonctionnalité en tant que microservices
- Le proxy achemine vers les anciens/nouveaux systèmes
2. Modèles de Communication
Synchrone (Requête/Réponse)
- API REST
- gRPC
- GraphQL
Asynchrone (Événements/Messages)
- Streaming d'événements (Kafka)
- Files d'attente (RabbitMQ, SQS)
- Modèles Pub/Sub
3. Gestion des Données
Base de Données Par Service
- Chaque service possède ses données
- Pas de bases de données partagées
- Couplage faible
Modèle Saga
- Transactions distribuées
- Actions de compensation
- Cohérence éventuelle
4. Modèles de Résilience
Circuit Breaker
- Échouer rapidement en cas d'erreurs répétées
- Prévenir les défaillances en cascade
Retry avec Backoff
- Gestion des défaillances transitoires
- Backoff exponentiel
Bulkhead
- Isoler les ressources
- Limiter l'impact des défaillances
Modèles de Décomposition de Service
Modèle 1 : Par Capacité Métier
# Exemple e-commerce
# Order Service
class OrderService:
"""Gère le cycle de vie des commandes."""
async def create_order(self, order_data: dict) -> Order:
order = Order.create(order_data)
# Publier un événement pour les autres services
await self.event_bus.publish(
OrderCreatedEvent(
order_id=order.id,
customer_id=order.customer_id,
items=order.items,
total=order.total
)
)
return order
# Payment Service (service séparé)
class PaymentService:
"""Gère le traitement des paiements."""
async def process_payment(self, payment_request: PaymentRequest) -> PaymentResult:
# Traiter le paiement
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
# Inventory Service (service séparé)
class InventoryService:
"""Gère la gestion des stocks."""
async def reserve_items(self, order_id: str, items: List[OrderItem]) -> ReservationResult:
# Vérifier la disponibilité
for item in items:
available = await self.inventory_repo.get_available(item.product_id)
if available < item.quantity:
return ReservationResult(
success=False,
error=f"Insufficient inventory for {item.product_id}"
)
# Réserver les articles
reservation = await self.create_reservation(order_id, items)
await self.event_bus.publish(
InventoryReservedEvent(
order_id=order_id,
reservation_id=reservation.id
)
)
return ReservationResult(success=True, reservation=reservation)
Modèle 2 : API Gateway
from fastapi import FastAPI, HTTPException, Depends
import httpx
from circuitbreaker import circuit
app = FastAPI()
class APIGateway:
"""Point d'entrée central pour toutes les requêtes client."""
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):
"""Appeler le service de commande avec 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:
"""Agréger les données de plusieurs services."""
# Requêtes parallèles
order, payment, inventory = await asyncio.gather(
self.call_order_service(f"/orders/{order_id}"),
self.call_payment_service(f"/payments/order/{order_id}"),
self.call_inventory_service(f"/reservations/order/{order_id}"),
return_exceptions=True
)
# Gérer les défaillances partielles
result = {"order": order}
if not isinstance(payment, Exception):
result["payment"] = payment
if not isinstance(inventory, Exception):
result["inventory"] = inventory
return result
@app.post("/api/orders")
async def create_order(
order_data: dict,
gateway: APIGateway = Depends()
):
"""Point de terminaison API Gateway."""
try:
# Acheminer vers le service de commande
order = await gateway.call_order_service(
"/orders",
method="POST",
json=order_data
)
return {"order": order}
except httpx.HTTPError as e:
raise HTTPException(status_code=503, detail="Order service unavailable")
Modèles de Communication
Modèle 1 : Communication REST Synchrone
# Service A appelle Service B
import httpx
from tenacity import retry, stop_after_attempt, wait_exponential
class ServiceClient:
"""Client HTTP avec retries et 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 avec retries automatiques."""
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):
"""Requête POST."""
response = await self.client.post(f"{self.base_url}{path}", **kwargs)
response.raise_for_status()
return response.json()
# Utilisation
payment_client = ServiceClient("http://payment-service:8001")
result = await payment_client.post("/payments", json=payment_data)
Modèle 2 : Orienté Événements Asynchrone
# Communication orientée événements avec Kafka
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:
"""Publication et souscription d'événements."""
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):
"""Publier un événement vers un topic Kafka."""
topic = event.event_type
await self.producer.send_and_wait(
topic,
value=asdict(event),
key=event.aggregate_id.encode()
)
async def subscribe(self, topic: str, handler: callable):
"""S'abonner aux événements."""
consumer = AIOKafkaConsumer(
topic,
bootstrap_servers=self.bootstrap_servers,
value_deserializer=lambda v: json.loads(v.decode()),
group_id="my-service"
)
await consumer.start()
try:
async for message in consumer:
event_data = message.value
await handler(event_data)
finally:
await consumer.stop()
# Order Service publie un événement
async def create_order(order_data: dict):
order = await save_order(order_data)
event = DomainEvent(
event_id=str(uuid.uuid4()),
event_type="OrderCreated",
aggregate_id=order.id,
occurred_at=datetime.now(),
data={
"order_id": order.id,
"customer_id": order.customer_id,
"total": order.total
}
)
await event_bus.publish(event)
# Inventory Service écoute OrderCreated
async def handle_order_created(event_data: dict):
"""Réagir à la création de commande."""
order_id = event_data["data"]["order_id"]
items = event_data["data"]["items"]
# Réserver le stock
await reserve_inventory(order_id, items)
Modèle 3 : Modèle Saga (Transactions Distribuées)
# Orchestration Saga pour l'accomplissement des commandes
from enum import Enum
from typing import List, Callable
class SagaStep:
"""Étape unique dans une 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:
"""Saga orchestrée pour l'accomplissement des commandes."""
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=self.refund_payment
),
SagaStep(
"confirm_order",
action=self.confirm_order,
compensation=self.cancel_order_confirmation
)
]
async def execute(self, order_data: dict) -> SagaResult:
"""Exécuter les étapes de la saga."""
completed_steps = []
context = {"order_data": order_data}
try:
for step in self.steps:
# Exécuter l'étape
result = await step.action(context)
if not result.success:
# Compenser
await self.compensate(completed_steps, context)
return SagaResult(
status=SagaStatus.FAILED,
error=result.error
)
completed_steps.append(step)
context.update(result.data)
return SagaResult(status=SagaStatus.COMPLETED, data=context)
except Exception as e:
# Compenser en cas d'erreur
await self.compensate(completed_steps, context)
return SagaResult(status=SagaStatus.FAILED, error=str(e))
async def compensate(self, completed_steps: List[SagaStep], context: dict):
"""Exécuter les actions de compensation dans l'ordre inverse."""
for step in reversed(completed_steps):
try:
await step.compensation(context)
except Exception as e:
# Journaliser l'échec de la compensation
print(f"Compensation failed for {step.name}: {e}")
# Implémentations des étapes
async def create_order(self, context: dict) -> StepResult:
order = await order_service.create(context["order_data"])
return StepResult(success=True, data={"order_id": order.id})
async def cancel_order(self, context: dict):
await order_service.cancel(context["order_id"])
async def reserve_inventory(self, context: dict) -> StepResult:
result = await inventory_service.reserve(
context["order_id"],
context["order_data"]["items"]
)
return StepResult(
success=result.success,
data={"reservation_id": result.reservation_id}
)
async def release_inventory(self, context: dict):
await inventory_service.release(context["reservation_id"])
async def process_payment(self, context: dict) -> StepResult:
result = await payment_service.charge(
context["order_id"],
context["order_data"]["total"]
)
return StepResult(
success=result.success,
data={"transaction_id": result.transaction_id},
error=result.error
)
async def refund_payment(self, context: dict):
await payment_service.refund(context["transaction_id"])
Modèles de Résilience
Modèle Circuit Breaker
from enum import Enum
from datetime import datetime, timedelta
from typing import Callable, Any
class CircuitState(Enum):
CLOSED = "closed" # Fonctionnement normal
OPEN = "open" # Défaillance, rejeter les requêtes
HALF_OPEN = "half_open" # Tester la récupération
class CircuitBreaker:
"""Circuit breaker pour les appels de service."""
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:
"""Exécuter la fonction avec circuit breaker."""
if self.state == CircuitState.OPEN:
if self._should_attempt_reset():
self.state = CircuitState.HALF_OPEN
else:
raise CircuitBreakerOpenError("Circuit breaker is open")
try:
result = await func(*args, **kwargs)
self._on_success()
return result
except Exception as e:
self._on_failure()
raise
def _on_success(self):
"""Gérer l'appel réussi."""
self.failure_count = 0
if self.state == CircuitState.HALF_OPEN:
self.success_count += 1
if self.success_count >= self.success_threshold:
self.state = CircuitState.CLOSED
self.success_count = 0
def _on_failure(self):
"""Gérer l'appel échoué."""
self.failure_count += 1
if self.failure_count >= self.failure_threshold:
self.state = CircuitState.OPEN
self.opened_at = datetime.now()
if self.state == CircuitState.HALF_OPEN:
self.state = CircuitState.OPEN
self.opened_at = datetime.now()
def _should_attempt_reset(self) -> bool:
"""Vérifier si suffisamment de temps s'est écoulé pour réessayer."""
return (
datetime.now() - self.opened_at
> timedelta(seconds=self.recovery_timeout)
)
# Utilisation
breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=30)
async def call_payment_service(payment_data: dict):
return await breaker.call(
payment_client.process_payment,
payment_data
)