Message Queues - le système de messagers de l'Empire

Un légionnaire achève son inscription, et le service lui envoie une lettre de bienvenue. Le serveur de courrier se trouve justement hors service - et l'inscription échoue. L'homme reste rejeté devant la porte, alors que son entrée au registre a parfaitement réussi. Une affaire annexe a échoué et a entraîné l'affaire principale dans sa chute.

Rome a résolu cela autrement. Un commandant n'attend pas que le messager revienne de Gaule - il laisse la lettre dans la boîte des courriers et retourne à ses affaires. Le messager la prendra quand il pourra ; si le cheval boite, la lettre attendra. Dans les systèmes, nous appelons cette boîte une file de messages.

Ce qu'apporte une file

Il y a un seul avantage et c'est tout l'enjeu : la communication asynchrone. Le producteur - celui qui envoie - n'attend pas le consommateur, celui qui reçoit. Il laisse le message et poursuit.

Trois conséquences en découlent. L'inscription s'achève aussitôt, car elle n'est plus liée au courrier. La panne d'un consommateur ne renverse pas le producteur - les messages attendent dans la file qu'il se relève. Et quand les lettres s'accumulent, tu ajoutes d'autres consommateurs à la même file sans toucher au producteur.

Le producteur - déposer la lettre

L'application émettrice a besoin d'un client relié à un broker, le serveur qui gère les files :

1ClientsModule.register([
2  {
3    name: 'LEGION_SERVICE',
4    transport: Transport.RMQ,
5    options: {
6      urls: ['amqp://localhost:5672'],
7      queue: 'legion_queue',
8    },
9  },
10]);

Transport.RMQ désigne RabbitMQ - le broker le plus répandu. urls est l'adresse du serveur et queue le nom de la boîte où iront les lettres. Le nom LEGION_SERVICE servira à injecter le client là où il sera nécessaire.

emit contre send - deux sortes de lettres

Le client en main, nous envoyons un message par l'une de deux méthodes. La différence entre elles est fondamentale :

1// emit - envoie et oublie, nous n'attendons pas de réponse
2this.client.emit('legion.created', { id: 1, name: 'Legio X' });
3
4// send - envoie et attend une réponse
5const result = await this.client.send('legion.count', {}).toPromise();

emit() est du fire-and-forget - tu laisses la lettre et poursuis, sans attendre de réponse. C'est ainsi que tu annonces des faits : une légion a été créée, un paiement est passé, un utilisateur s'est inscrit. L'expéditeur se moque de ce que le destinataire en fera.

send() attend une réponse - c'est une question, non une annonce. Tu l'emploies quand il te faut un résultat : combien de légions existent, si cet identifiant existe.

Le choix entre les deux décide si tu profites de la file. send() rétablit l'attente, et avec elle le couplage que nous fuyions - le consommateur doit de nouveau tourner pour que le producteur avance. Alors pour les notifications, les e-mails et les journaux, toujours emit() ; garde send() pour les cas où tu as vraiment besoin d'une réponse.

Le consommateur - recevoir la lettre

De l'autre côté de la file se tient un service à l'écoute :

1@Controller()
2export class LegionConsumer {
3  @EventPattern('legion.created')
4  async handleLegionCreated(@Payload() data: LegionCreatedDto) {
5    await this.mailService.sendWelcome(data);
6  }
7}

@EventPattern('legion.created') dit : cette méthode traite les messages portant ce nom. Le nom est un contrat entre producteur et consommateur - il doit correspondre caractère pour caractère à celui passé à emit(). @Payload() extrait le corps du message, c'est-à-dire l'objet passé en second argument.

Note que le consommateur ressemble à un contrôleur ordinaire - parce qu'il fonctionne à peu près pareil. La différence tient à l'origine de la requête : non pas le réseau HTTP, mais la file.

L'accusé de réception

Le broker doit savoir si la lettre est arrivée et a été traitée. Sans cette information, il ne peut pas la supprimer - et s'il la supprimait dès l'envoi, la panne d'un consommateur en pleine tâche signifierait la perte définitive du message.

D'où l'accusé de réception, l'acknowledgement :

1@EventPattern('legion.created')
2async handleLegionCreated(@Payload() data: LegionCreatedDto, @Ctx() context: RmqContext) {
3  const channel = context.getChannelRef();
4  const originalMsg = context.getMessage();
5
6  try {
7    await this.mailService.sendWelcome(data);
8
9    channel.ack(originalMsg);
10  } catch (error) {
11    channel.nack(originalMsg, false, true);
12  }
13}

@Ctx() donne accès au contexte RabbitMQ. getChannelRef() renvoie le canal de communication avec le broker, et getMessage() le message brut que l'accusé concerne.

L'essentiel est dans le bloc try. channel.ack(originalMsg) confirme le traitement réussi - c'est seulement alors que le broker retire le message de la file. Note l'emplacement de cet appel : après l'exécution du travail, non avant. Confirmer en tête de méthode signifierait « j'ai reçu » et non « j'ai traité », et un échec d'envoi de l'e-mail se solderait par la perte silencieuse de la lettre.

Dans le catch se trouve nack - un refus de confirmer. Son dernier argument, true, demande au broker de remettre le message dans la file pour réessayer plus tard. C'est précisément la résistance pour laquelle nous avons introduit les files : une tentative ratée ne détruit pas le message.

Récapitulatif

La boîte des courriers fonctionne, les lettres arrivent :

• une file de messages apporte la communication asynchrone - le producteur n'attend pas le consommateur,
• ainsi la panne d'un consommateur ne renverse pas le producteur, et les messages attendent le retour du destinataire,
• nous configurons le client par ClientsModule.register avec Transport.RMQ, l'adresse du broker et un nom de file,
• emit() est du fire-and-forget - tu annonces un fait sans attendre de réponse ; send() attend une réponse et rétablit donc le couplage,
• utilise emit() pour les notifications et les e-mails ; send() seulement quand un résultat t'est vraiment nécessaire,
• le consommateur écoute par @EventPattern('nom') et extrait le corps par @Payload() ; le nom doit correspondre à celui d'emit(),
• channel.ack(originalMsg) confirme le traitement, et c'est seulement alors que le broker retire le message de la file,
• confirme après le travail, non avant - sinon une panne signifie un message perdu en silence,
• nack avec un dernier argument true remet le message dans la file pour une nouvelle tentative,
• le flux complet : le producteur envoie par emit → le message attend dans la file → le consommateur le reçoit par @EventPattern → il confirme par channel.ack.

C'est la dernière leçon de ce module. Tu sais désormais vérifier chaque pièce isolément, leur coopération, la route entière d'une requête, mesurer la couverture et l'endurance, et maintenant aussi séparer les services pour qu'une panne n'entraîne pas les autres. En attendant, retiens ceci : une file est une boîte de courriers - le commandant y laisse sa lettre et retourne à ses affaires, et l'accusé de réception n'arrive que lorsque le messager est vraiment parvenu à destination.

1// Message Queues - Le système de messagers de l'Empire romain
2import { Injectable } from '@nestjs/common';
3
4// 1. Implémentation simple d'une file de messages
5class MessageQueue<T> {
6  private queue: Array<{ id: string; data: T; timestamp: Date }> = [];
7  private handlers: Map<string, (data: T) => Promise<void>> = new Map();
8
9  // Envoyer un message dans la file
10  async publish(channel: string, data: T): Promise<string> {
11    const id = `msg-${Date.now()}`;
12    this.queue.push({ id, data, timestamp: new Date() });
13
14    const handler = this.handlers.get(channel);
15    if (handler) {
16      await handler(data);
17    }
18    return id;
19  }
20
21  // S'abonner à un canal
22  subscribe(channel: string, handler: (data: T) => Promise<void>) {
23    this.handlers.set(channel, handler);
24  }
25
26  getQueueStatus() {
27    return {
28      size: this.queue.length,
29      channels: Array.from(this.handlers.keys()),
30      oldestMessage: this.queue[0]?.timestamp,
31    };
32  }
33}
34
35// 2. NestJS Bull Queue - configuration
36// @Module({
37//   imports: [
38//     BullModule.forRoot({
39//       redis: { host: 'localhost', port: 6379 },
40//     }),
41//     BullModule.registerQueue({ name: 'tribute-collection' }),
42//   ],
43// })
44// export class QueueModule {}
45
46// 3. Producer - ajouter des jobs à la file
47@Injectable()
48class TributeQueueService {
49  private queue = new MessageQueue<any>();
50
51  constructor() {
52    this.queue.subscribe('tribute', async (data) => {
53      console.log(
54        `Collecting tribute: ${data.province} - ${data.amount} gold`
55      );
56    });
57
58    this.queue.subscribe('alert', async (data) => {
59      console.log(`ALERT: ${data.message}`);
60    });
61  }
62
63  async scheduleTributeCollection(province: string, amount: number) {
64    return this.queue.publish('tribute', { province, amount });
65  }
66
67  async sendAlert(message: string, priority: 'low' | 'high') {
68    return this.queue.publish('alert', { message, priority });
69  }
70
71  getStatus() {
72    return this.queue.getQueueStatus();
73  }
74}
75
76// 4. Patterns de communication :
77// - Point-to-Point : producteur -> consommateur
78// - Pub/Sub : producteur -> plusieurs consommateurs
79// - Request/Reply : le producteur attend une réponse
80// - Fan-out : message vers plusieurs files
81
82// 5. Processor - traitement des jobs
83// @Processor('tribute-collection')
84// class TributeProcessor {
85//   @Process('collect')
86//   async handleCollect(job: Job<TributeData>) {
87//     const { province, amount } = job.data;
88//     return { collected: true, province };
89//   }
90//
91//   @OnQueueCompleted()
92//   onCompleted(job: Job, result: any) {
93//     console.log(`Job ${job.id} completed`);
94//   }
95// }
96