Używamy cookies, żeby zwiększyć Twoje doświadczenia na stronie
Message Queues - System posłańców Imperium
Witaj, architekcie systemu! Konsul Caesar.js wprowadza Cię teraz w świat Message Queues - systemu komunikacji między serwisami, który działa jak armia posłańców Imperium Rzymskiego. Gdy jeden legion potrzebuje przekazać rozkaz innemu, wysyła posłańca - wiadomość trafia do kolejki i czeka na odbiorcę.
Czym są Message Queues?
Message Queues to mechanizm asynchronicznej komunikacji między serwisami. Zamiast bezpośredniego połączenia (synchronicznego), producent wysyła wiadomość do kolejki, a konsument odbiera ją gdy jest gotowy.
Analogia rzymska
1Komunikacja SYNCHRONICZNA (bez kolejki):
2┌─────────┐ ┌─────────┐
3│ Legat │ ──────► │ Centurion│
4│ (czeka) │ ◄────── │ │
5└─────────┘ └─────────┘
6Legat musi czekać na odpowiedź!
7
8Komunikacja ASYNCHRONICZNA (z kolejką):
9┌─────────┐ ┌─────────┐ ┌─────────────┐
10│ Legat │ ──►│ Posłaniec│ ──►│ Centurion │
11│ (pracuje│ │ (kolejka)│ │ (przetwarza)│
12│ dalej) │ └─────────┘ └─────────────┘
13└─────────┘
14Legat może robić inne rzeczy!RabbitMQ - Podstawy
RabbitMQ to jeden z najpopularniejszych message brokerów. Implementuje protokół AMQP (Advanced Message Queuing Protocol).
Instalacja i konfiguracja w NestJS
1# Instalacja pakietów
2npm install @nestjs/microservices amqplib amqp-connection-manager
3
4# Docker dla RabbitMQ
5docker run -d --hostname rabbit --name rabbitmq \
6 -p 5672:5672 -p 15672:15672 \
7 rabbitmq:3-managementKonfiguracja modułu
1// app.module.ts
2import { Module } from '@nestjs/common';
3import { ClientsModule, Transport } from '@nestjs/microservices';
4
5@Module({
6 imports: [
7 ClientsModule.register([
8 {
9 name: 'ORDERS_SERVICE',
10 transport: Transport.RMQ,
11 options: {
12 urls: ['amqp://localhost:5672'],
13 queue: 'orders_queue',
14 queueOptions: {
15 durable: true, // Kolejka przetrwa restart brokera
16 },
17 },
18 },
19 ]),
20 ],
21})
22export class AppModule {}Producent wiadomości
1// orders/orders.service.ts
2import { Injectable, Inject } from '@nestjs/common';
3import { ClientProxy } from '@nestjs/microservices';
4
5interface OrderCreatedEvent {
6 orderId: string;
7 items: Array<{ productId: string; quantity: number }>;
8 totalAmount: number;
9 userId: string;
10 timestamp: Date;
11}
12
13@Injectable()
14export class OrdersService {
15 constructor(
16 @Inject('ORDERS_SERVICE') private readonly client: ClientProxy,
17 ) {}
18
19 async createOrder(orderData: CreateOrderDto): Promise<Order> {
20 // 1. Zapisz zamówienie w bazie
21 const order = await this.ordersRepository.create(orderData);
22
23 // 2. Wyślij event do kolejki (asynchronicznie!)
24 const event: OrderCreatedEvent = {
25 orderId: order.id,
26 items: order.items,
27 totalAmount: order.total,
28 userId: order.userId,
29 timestamp: new Date(),
30 };
31
32 // emit() - fire-and-forget (nie czeka na odpowiedź)
33 this.client.emit('order_created', event);
34
35 // Możemy też użyć send() jeśli potrzebujemy odpowiedzi
36 // const result = await firstValueFrom(
37 // this.client.send('process_order', event)
38 // );
39
40 return order;
41 }
42
43 async cancelOrder(orderId: string): Promise<void> {
44 const order = await this.ordersRepository.findById(orderId);
45 order.status = 'cancelled';
46 await this.ordersRepository.save(order);
47
48 // Powiadom inne serwisy o anulowaniu
49 this.client.emit('order_cancelled', {
50 orderId,
51 reason: 'User requested cancellation',
52 timestamp: new Date(),
53 });
54 }
55}Konsument wiadomości
1// notifications/notifications.controller.ts
2import { Controller } from '@nestjs/common';
3import { EventPattern, Payload, Ctx, RmqContext } from '@nestjs/microservices';
4
5@Controller()
6export class NotificationsController {
7 constructor(private readonly notificationService: NotificationService) {}
8
9 @EventPattern('order_created')
10 async handleOrderCreated(
11 @Payload() data: OrderCreatedEvent,
12 @Ctx() context: RmqContext,
13 ): Promise<void> {
14 const channel = context.getChannelRef();
15 const originalMsg = context.getMessage();
16
17 try {
18 console.log('Otrzymano event order_created:', data.orderId);
19
20 // Wyślij powiadomienie email
21 await this.notificationService.sendOrderConfirmation(data);
22
23 // Wyślij powiadomienie push
24 await this.notificationService.sendPushNotification(
25 data.userId,
26 'Zamówienie przyjęte!',
27 \`Twoje zamówienie #\${data.orderId} zostało przyjęte.\`
28 );
29
30 // Potwierdź przetworzenie wiadomości (ACK)
31 channel.ack(originalMsg);
32 } catch (error) {
33 console.error('Błąd przetwarzania:', error);
34
35 // Odrzuć wiadomość i wrzuć z powrotem do kolejki (NACK)
36 channel.nack(originalMsg, false, true);
37 }
38 }
39
40 @EventPattern('order_cancelled')
41 async handleOrderCancelled(
42 @Payload() data: OrderCancelledEvent,
43 @Ctx() context: RmqContext,
44 ): Promise<void> {
45 const channel = context.getChannelRef();
46 const originalMsg = context.getMessage();
47
48 try {
49 await this.notificationService.sendCancellationEmail(data);
50 channel.ack(originalMsg);
51 } catch (error) {
52 channel.nack(originalMsg, false, true);
53 }
54 }
55}Zaawansowane wzorce Message Queues
1. Dead Letter Queue (DLQ)
Kolejka dla wiadomości, które nie mogły być przetworzone:
1// config/rabbitmq.config.ts
2export const rabbitMQConfig = {
3 queue: 'orders_queue',
4 queueOptions: {
5 durable: true,
6 arguments: {
7 'x-dead-letter-exchange': 'dlx.exchange',
8 'x-dead-letter-routing-key': 'dlx.orders',
9 'x-message-ttl': 86400000, // 24h
10 'x-max-retries': 3,
11 },
12 },
13};
14
15// Dead Letter Queue handler
16@Controller()
17export class DeadLetterHandler {
18 @EventPattern('dlx.orders')
19 async handleDeadLetter(
20 @Payload() data: any,
21 @Ctx() context: RmqContext,
22 ): Promise<void> {
23 const channel = context.getChannelRef();
24 const originalMsg = context.getMessage();
25
26 // Loguj nieudane wiadomości do monitoringu
27 await this.alertService.sendAlert({
28 type: 'DEAD_LETTER',
29 queue: 'orders_queue',
30 message: data,
31 headers: originalMsg.properties.headers,
32 });
33
34 // Zapisz do bazy dla późniejszej analizy
35 await this.failedMessagesRepository.save({
36 payload: data,
37 error: originalMsg.properties.headers['x-death'],
38 createdAt: new Date(),
39 });
40
41 channel.ack(originalMsg);
42 }
43}2. Retry z Exponential Backoff
1// common/retry.decorator.ts
2import { Injectable } from '@nestjs/common';
3
4@Injectable()
5export class MessageRetryService {
6 private readonly maxRetries = 5;
7 private readonly baseDelay = 1000; // 1 sekunda
8
9 async processWithRetry<T>(
10 handler: () => Promise<T>,
11 context: RmqContext,
12 ): Promise<T> {
13 const channel = context.getChannelRef();
14 const originalMsg = context.getMessage();
15 const retryCount = this.getRetryCount(originalMsg);
16
17 try {
18 const result = await handler();
19 channel.ack(originalMsg);
20 return result;
21 } catch (error) {
22 if (retryCount < this.maxRetries) {
23 // Oblicz delay z exponential backoff
24 const delay = this.baseDelay * Math.pow(2, retryCount);
25
26 console.log(\`Retry \${retryCount + 1}/\${this.maxRetries} za \${delay}ms\`);
27
28 // Odrzuć i zaplanuj ponowne przetworzenie
29 await this.scheduleRetry(originalMsg, delay, retryCount + 1);
30 channel.ack(originalMsg); // ACK oryginał, nowy trafi do kolejki
31 } else {
32 // Przekroczyliśmy limit - wyślij do DLQ
33 channel.nack(originalMsg, false, false);
34 }
35 throw error;
36 }
37 }
38
39 private getRetryCount(msg: any): number {
40 return msg.properties.headers?.['x-retry-count'] || 0;
41 }
42
43 private async scheduleRetry(
44 originalMsg: any,
45 delay: number,
46 retryCount: number,
47 ): Promise<void> {
48 // Użyj delayed message plugin lub osobnej kolejki z TTL
49 await this.rabbitClient.emit('orders_queue_delayed', {
50 payload: JSON.parse(originalMsg.content.toString()),
51 headers: {
52 'x-retry-count': retryCount,
53 'x-delay': delay,
54 },
55 });
56 }
57}3. Saga Pattern - transakcje rozproszone
1// sagas/order.saga.ts
2import { Injectable } from '@nestjs/common';
3
4interface SagaStep {
5 name: string;
6 execute: () => Promise<void>;
7 compensate: () => Promise<void>;
8}
9
10@Injectable()
11export class OrderSaga {
12 private executedSteps: SagaStep[] = [];
13
14 async execute(orderId: string): Promise<void> {
15 const steps: SagaStep[] = [
16 {
17 name: 'reserve_inventory',
18 execute: () => this.inventoryService.reserve(orderId),
19 compensate: () => this.inventoryService.release(orderId),
20 },
21 {
22 name: 'process_payment',
23 execute: () => this.paymentService.charge(orderId),
24 compensate: () => this.paymentService.refund(orderId),
25 },
26 {
27 name: 'create_shipment',
28 execute: () => this.shippingService.createShipment(orderId),
29 compensate: () => this.shippingService.cancelShipment(orderId),
30 },
31 {
32 name: 'send_confirmation',
33 execute: () => this.notificationService.sendConfirmation(orderId),
34 compensate: () => this.notificationService.sendCancellation(orderId),
35 },
36 ];
37
38 try {
39 for (const step of steps) {
40 console.log(\`Executing step: \${step.name}\`);
41 await step.execute();
42 this.executedSteps.push(step);
43 }
44
45 // Wszystkie kroki wykonane - emituj sukces
46 this.eventBus.emit('order_saga_completed', { orderId });
47 } catch (error) {
48 console.error('Saga failed, starting compensation...');
49 await this.compensate();
50
51 this.eventBus.emit('order_saga_failed', {
52 orderId,
53 error: error.message,
54 failedStep: this.executedSteps.length,
55 });
56
57 throw error;
58 }
59 }
60
61 private async compensate(): Promise<void> {
62 // Wykonaj kompensację w odwrotnej kolejności
63 for (const step of this.executedSteps.reverse()) {
64 try {
65 console.log(\`Compensating step: \${step.name}\`);
66 await step.compensate();
67 } catch (compensationError) {
68 console.error(\`Compensation failed for \${step.name}:\`, compensationError);
69 // Log do alertów - wymaga interwencji manualnej
70 }
71 }
72 }
73}Apache Kafka - dla dużej skali
Kafka to platforma do streamingu danych, idealna dla dużych wolumenów:
1// kafka/kafka.module.ts
2import { Module } from '@nestjs/common';
3import { ClientsModule, Transport } from '@nestjs/microservices';
4
5@Module({
6 imports: [
7 ClientsModule.register([
8 {
9 name: 'KAFKA_SERVICE',
10 transport: Transport.KAFKA,
11 options: {
12 client: {
13 clientId: 'orders-service',
14 brokers: ['localhost:9092'],
15 },
16 consumer: {
17 groupId: 'orders-consumer-group',
18 },
19 },
20 },
21 ]),
22 ],
23})
24export class KafkaModule {}
25
26// Producent Kafka
27@Injectable()
28export class EventProducerService {
29 constructor(
30 @Inject('KAFKA_SERVICE') private readonly kafka: ClientProxy,
31 ) {}
32
33 async publishOrderEvent(event: OrderEvent): Promise<void> {
34 await this.kafka.emit('orders.events', {
35 key: event.orderId, // Partitioning key
36 value: event,
37 headers: {
38 'event-type': event.type,
39 'event-version': '1.0',
40 'correlation-id': event.correlationId,
41 },
42 });
43 }
44}
45
46// Konsument Kafka
47@Controller()
48export class OrderEventsConsumer {
49 @EventPattern('orders.events')
50 async handleOrderEvent(
51 @Payload() message: any,
52 @Ctx() context: KafkaContext,
53 ): Promise<void> {
54 const { offset } = context.getMessage();
55 const partition = context.getPartition();
56 const topic = context.getTopic();
57
58 console.log(\`Processing message from \${topic}[\${partition}] offset \${offset}\`);
59
60 // Przetwórz event
61 await this.processEvent(message.value);
62
63 // Kafka automatycznie commituje offset po udanym przetworzeniu
64 }
65}Monitorowanie kolejek
1// monitoring/queue-monitor.service.ts
2@Injectable()
3export class QueueMonitorService {
4 private readonly metrics = new Map<string, QueueMetrics>();
5
6 @Cron('*/30 * * * * *') // Co 30 sekund
7 async collectMetrics(): Promise<void> {
8 const queues = ['orders_queue', 'notifications_queue', 'payments_queue'];
9
10 for (const queue of queues) {
11 const stats = await this.rabbitMQAdmin.getQueueStats(queue);
12
13 const metrics: QueueMetrics = {
14 queue,
15 messageCount: stats.messages,
16 consumerCount: stats.consumers,
17 publishRate: stats.message_stats?.publish_details?.rate || 0,
18 deliverRate: stats.message_stats?.deliver_details?.rate || 0,
19 timestamp: new Date(),
20 };
21
22 this.metrics.set(queue, metrics);
23
24 // Alert jeśli kolejka rośnie zbyt szybko
25 if (metrics.messageCount > 10000) {
26 await this.alertService.send({
27 severity: 'WARNING',
28 message: \`Queue \${queue} has \${metrics.messageCount} pending messages\`,
29 });
30 }
31
32 // Alert jeśli brak konsumentów
33 if (metrics.consumerCount === 0) {
34 await this.alertService.send({
35 severity: 'CRITICAL',
36 message: \`Queue \${queue} has no active consumers!\`,
37 });
38 }
39 }
40 }
41}Najlepsze praktyki
1. Idempotentność
1// Zawsze sprawdzaj czy wiadomość nie była już przetworzona
2@EventPattern('order_created')
3async handleOrder(@Payload() data: OrderEvent): Promise<void> {
4 const alreadyProcessed = await this.cache.get(\`processed:\${data.eventId}\`);
5
6 if (alreadyProcessed) {
7 console.log(\`Event \${data.eventId} already processed, skipping\`);
8 return;
9 }
10
11 await this.processOrder(data);
12 await this.cache.set(\`processed:\${data.eventId}\`, true, 86400); // 24h TTL
13}2. Wersjonowanie wiadomości
1interface MessageEnvelope<T> {
2 version: string;
3 timestamp: Date;
4 correlationId: string;
5 payload: T;
6}
7
8// Handler obsługujący różne wersje
9@EventPattern('order_created')
10async handleOrder(@Payload() envelope: MessageEnvelope<any>): Promise<void> {
11 switch (envelope.version) {
12 case '1.0':
13 await this.handleV1(envelope.payload);
14 break;
15 case '2.0':
16 await this.handleV2(envelope.payload);
17 break;
18 default:
19 throw new Error(\`Unsupported message version: \${envelope.version}\`);
20 }
21}Message Queues to fundament skalowalnych systemów rozproszonych. Tak jak posłańcy Imperium Rzymskiego zapewniali komunikację między legionami na ogromnych odległościach, tak kolejki wiadomości zapewniają niezawodną komunikację między mikroserwisami!