Architecture orientée événements : comment nous avons remplacé un système monolithique qui coûtait 50 000 dollars par mois

Le problème : un système monolithique incapable de s'adapter à la croissance

Avant : La chaîne synchrone de la douleur

# L'ancienne méthode : tout est synchrone, tout est 
couplédéf process_order(request) :
    order = Order.objects.create(**request.data)    

 # Si l'une de ces opérations échoue, la commande échoue
    inventory_service.reserve(order)        # 200 ms
    payment_service.charge(order)           # 800 ms
    email_service.send_confirmation(order)  # 500 ms
    analytics_service.track(order)          # 300 ms
    warehouse_service.notify(order)         # 400 ms    

 # Total : temps de réponse 
minimum de 2 200 ms    # Si le service de messagerie est indisponible : la commande échoue
    # Si le service d'analyse est lent : le paiement est lent
    return Response({'order_id': order.id})

L'architecture : événements et commandes

Après : Approche orientée événements avec Celery + Redis

# La nouvelle approche : synchronisation pour le chemin critique, événements pour le 
reste def process_order(request):    
 order = Order.objects.create(**request.data)    

 # Chemin critique : doit aboutir (synchrone)    
 inventory_service.reserve(order)   # 200 ms    
 payment_service.charge(order)      # 800 ms    

 # Publication d'événement : « fire-and-forget » (asynchrone)
    publish_event('order.completed', {
        'order_id': order.id,
        'customer_email': order.customer.email,
        'total': str(order.total),
    })    

 # Total : temps de réponse de
 1000 ms    # Problème de messagerie ? La commande aboutit quand même. Nouvelle tentative plus tard.
    return Response({'order_id': order.id})


# Gestionnaires d'événements (workers séparés)
@celery_app.task(bind=True, max_retries=5, default_retry_delay=60)
def handle_order_completed(self, event_data):
    """Chaque gestionnaire est indépendant et peut être relancé"""
    try:
        email_service.send_confirmation(event_data)
    except Exception as exc:
        self.retry(exc=exc, countdown=2 ** self.request.retries * 60)

Le modèle de bus d'événements

import redisimport
 
jsonfrom datetime import 

datetimeclass EventBus:
    def __init__(self):
        self.redis = redis.Redis(host='redis', port=6379, db=0)    

 def publish(self, event_type: str, data: dict):
        """Publier un événement sur Redis Stream"""
        event = {
            'type': event_type,
            'data': json.dumps(data),
            'timestamp': datetime.utcnow().isoformat(),
        }
        self.redis.xadd(f'events:{event_type}', event)    

 def subscribe(self, event_type: str, group: str, consumer: str):
        """Lire les événements avec un groupe de consommateurs (chaque événement traité une seule fois)"""
        try:
            self.redis.xgroup_create(
                f'events:{event_type}', group, id='0', mkstream=True
            )
        except redis.ResponseError:
            pass  # Le groupe existe déjà        

 while True:
            events = self.redis.xreadgroup(
                group, consumer,
                {f'events:{event_type}': '>'},
                count=10, block=5000
            )
            for stream, messages in events:
                for msg_id, fields in messages:
                    yield msg_id, json.loads(fields[b'data'])
                    self.redis.xack(stream, group, msg_id)

Résultats après 3 mois

Quand NE PAS utiliser une architecture orientée événements

Commencez par un monolithe. Définissez des événements lorsque vous rencontrez des difficultés. La pire architecture est celle conçue pour résoudre des problèmes que vous n'avez pas encore.

— alokknight Ingénierie