Chapitre 3: Messaging Asynchrone avec RabbitMQ

Objectifs du Chapitre

• Comprendre les concepts de messaging asynchrone
• Configurer RabbitMQ avec Spring Boot
• Implémenter des producteurs et consommateurs
• Gérer les exchanges, queues et bindings
• Implémenter des patterns de messaging

---

1. Introduction au Messaging

1.1 Pourquoi le Messaging Asynchrone

Synchrone (REST):
Service A --HTTP--> Service B --HTTP--> Service C
    └── Attend réponse ──┘     └── Attend réponse ──┘
    
Problèmes:
- Couplage temporel
- Service B down = Service A échoue
- Latence cumulée

Asynchrone (Messaging):
Service A --publish--> [Queue] --consume--> Service B
    └── Continue immédiatement

Avantages:
- Découplage
- Résilience
- Scalabilité
- Peak handling

1.2 Concepts RabbitMQ

Concept	Description
Producer	Envoie des messages
Consumer	Reçoit des messages
Queue	File d'attente des messages
Exchange	Route les messages vers les queues
Binding	Règle de routage exchange -> queue
Routing Key	Clé pour le routage

1.3 Types d'Exchange

Direct Exchange:
Producer --> [Exchange] --> Queue (routing_key exact match)

Fanout Exchange:
Producer --> [Exchange] --> Queue1
                      └--> Queue2
                      └--> Queue3

Topic Exchange:
Producer --> [Exchange] --> Queue (pattern: order.*, *.created)

Headers Exchange:
Producer --> [Exchange] --> Queue (header matching)

---

2. Configuration Spring AMQP

2.1 Dépendances Maven

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2.2 Configuration application.yml

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtual-host: /
    
    # Listener configuration
    listener:
      simple:
        acknowledge-mode: auto
        concurrency: 3
        max-concurrency: 10
        prefetch: 10
        retry:
          enabled: true
          initial-interval: 1000
          max-attempts: 3
          max-interval: 10000
          multiplier: 2
    
    # Template configuration
    template:
      retry:
        enabled: true
        initial-interval: 1000
        max-attempts: 3

2.3 Configuration Java

package com.example.messaging.config;

import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMQConfig {
    
    // Noms des queues et exchanges
    public static final String ORDER_QUEUE = "order.queue";
    public static final String ORDER_EXCHANGE = "order.exchange";
    public static final String ORDER_ROUTING_KEY = "order.created";
    
    public static final String NOTIFICATION_QUEUE = "notification.queue";
    public static final String NOTIFICATION_EXCHANGE = "notification.exchange";
    
    // Queue pour les commandes
    @Bean
    public Queue orderQueue() {
        return QueueBuilder.durable(ORDER_QUEUE)
            .withArgument("x-dead-letter-exchange", "dlx.exchange")
            .withArgument("x-dead-letter-routing-key", "dlx.order")
            .build();
    }
    
    // Exchange Direct pour les commandes
    @Bean
    public DirectExchange orderExchange() {
        return new DirectExchange(ORDER_EXCHANGE);
    }
    
    // Binding order queue <-> exchange
    @Bean
    public Binding orderBinding(Queue orderQueue, DirectExchange orderExchange) {
        return BindingBuilder
            .bind(orderQueue)
            .to(orderExchange)
            .with(ORDER_ROUTING_KEY);
    }
    
    // Queue pour les notifications
    @Bean
    public Queue notificationQueue() {
        return QueueBuilder.durable(NOTIFICATION_QUEUE).build();
    }
    
    // Exchange Fanout pour les notifications (broadcast)
    @Bean
    public FanoutExchange notificationExchange() {
        return new FanoutExchange(NOTIFICATION_EXCHANGE);
    }
    
    // Binding notification
    @Bean
    public Binding notificationBinding(Queue notificationQueue, 
            FanoutExchange notificationExchange) {
        return BindingBuilder
            .bind(notificationQueue)
            .to(notificationExchange);
    }
    
    // Convertisseur JSON
    @Bean
    public MessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }
    
    // Template configuré
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        template.setMessageConverter(jsonMessageConverter());
        return template;
    }
}

---

3. Producer (Envoi de Messages)

3.1 Message DTO

package com.example.messaging.dto;

import java.io.Serializable;
import java.time.LocalDateTime;

public class OrderEvent implements Serializable {
    
    private Long orderId;
    private Long userId;
    private String productName;
    private Double totalAmount;
    private String status;
    private LocalDateTime createdAt;
    
    public OrderEvent() {
        this.createdAt = LocalDateTime.now();
    }
    
    public OrderEvent(Long orderId, Long userId, String productName, 
            Double totalAmount, String status) {
        this();
        this.orderId = orderId;
        this.userId = userId;
        this.productName = productName;
        this.totalAmount = totalAmount;
        this.status = status;
    }
    
    // Getters et Setters
    public Long getOrderId() { return orderId; }
    public void setOrderId(Long orderId) { this.orderId = orderId; }
    
    public Long getUserId() { return userId; }
    public void setUserId(Long userId) { this.userId = userId; }
    
    public String getProductName() { return productName; }
    public void setProductName(String productName) { this.productName = productName; }
    
    public Double getTotalAmount() { return totalAmount; }
    public void setTotalAmount(Double totalAmount) { this.totalAmount = totalAmount; }
    
    public String getStatus() { return status; }
    public void setStatus(String status) { this.status = status; }
    
    public LocalDateTime getCreatedAt() { return createdAt; }
    public void setCreatedAt(LocalDateTime createdAt) { this.createdAt = createdAt; }
    
    @Override
    public String toString() {
        return "OrderEvent{orderId=" + orderId + ", userId=" + userId + 
               ", status=" + status + "}";
    }
}

3.2 Service Producer

package com.example.messaging.service;

import com.example.messaging.config.RabbitMQConfig;
import com.example.messaging.dto.OrderEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;

@Service
public class OrderEventProducer {
    
    private static final Logger log = LoggerFactory.getLogger(OrderEventProducer.class);
    
    private final RabbitTemplate rabbitTemplate;
    
    public OrderEventProducer(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }
    
    public void sendOrderCreatedEvent(OrderEvent event) {
        log.info("Sending order created event: {}", event);
        
        rabbitTemplate.convertAndSend(
            RabbitMQConfig.ORDER_EXCHANGE,
            RabbitMQConfig.ORDER_ROUTING_KEY,
            event
        );
        
        log.info("Order event sent successfully");
    }
    
    public void sendNotification(String message) {
        log.info("Broadcasting notification: {}", message);
        
        // Fanout - pas de routing key nécessaire
        rabbitTemplate.convertAndSend(
            RabbitMQConfig.NOTIFICATION_EXCHANGE,
            "",
            message
        );
    }
    
    // Envoi avec confirmation
    public void sendWithConfirmation(OrderEvent event) {
        rabbitTemplate.invoke(operations -> {
            operations.convertAndSend(
                RabbitMQConfig.ORDER_EXCHANGE,
                RabbitMQConfig.ORDER_ROUTING_KEY,
                event
            );
            
            // Attendre la confirmation
            boolean confirmed = operations.waitForConfirms(5000);
            if (!confirmed) {
                throw new RuntimeException("Message not confirmed by broker");
            }
            
            return null;
        });
    }
}

3.3 Utilisation dans un Service

@Service
public class OrderService {
    
    private final OrderRepository orderRepository;
    private final OrderEventProducer eventProducer;
    
    public OrderService(OrderRepository orderRepository, 
            OrderEventProducer eventProducer) {
        this.orderRepository = orderRepository;
        this.eventProducer = eventProducer;
    }
    
    @Transactional
    public Order createOrder(OrderRequest request) {
        // 1. Créer la commande
        Order order = new Order();
        order.setUserId(request.getUserId());
        order.setProductName(request.getProductName());
        order.setTotalAmount(request.getTotalAmount());
        order.setStatus("CREATED");
        
        Order saved = orderRepository.save(order);
        
        // 2. Publier l'événement
        OrderEvent event = new OrderEvent(
            saved.getId(),
            saved.getUserId(),
            saved.getProductName(),
            saved.getTotalAmount(),
            saved.getStatus()
        );
        
        eventProducer.sendOrderCreatedEvent(event);
        
        return saved;
    }
}

---

4. Consumer (Réception de Messages)

4.1 Listener Simple

package com.example.messaging.listener;

import com.example.messaging.config.RabbitMQConfig;
import com.example.messaging.dto.OrderEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class OrderEventListener {
    
    private static final Logger log = LoggerFactory.getLogger(OrderEventListener.class);
    
    @RabbitListener(queues = RabbitMQConfig.ORDER_QUEUE)
    public void handleOrderEvent(OrderEvent event) {
        log.info("Received order event: {}", event);
        
        try {
            // Traitement du message
            processOrder(event);
            log.info("Order event processed successfully");
        } catch (Exception e) {
            log.error("Error processing order event: {}", e.getMessage());
            throw e; // Re-throw pour retry/DLQ
        }
    }
    
    private void processOrder(OrderEvent event) {
        // Logique métier
        // - Envoyer email de confirmation
        // - Mettre à jour le stock
        // - Notifier le système de livraison
        log.info("Processing order {} for user {}", 
            event.getOrderId(), event.getUserId());
    }
}

4.2 Listener avec Accusé de Réception Manuel

@Component
public class OrderEventListener {
    
    @RabbitListener(queues = RabbitMQConfig.ORDER_QUEUE, 
                    ackMode = "MANUAL")
    public void handleOrderEvent(OrderEvent event, Channel channel,
            @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        
        log.info("Received order event: {}", event);
        
        try {
            processOrder(event);
            
            // Acknowledge success
            channel.basicAck(deliveryTag, false);
            log.info("Message acknowledged");
            
        } catch (Exception e) {
            log.error("Error processing: {}", e.getMessage());
            
            try {
                // Reject and requeue (ou envoyer en DLQ)
                channel.basicNack(deliveryTag, false, false);
            } catch (IOException ioEx) {
                log.error("Error rejecting message", ioEx);
            }
        }
    }
}

4.3 Listener pour Notifications (Fanout)

@Component
public class NotificationListener {
    
    private static final Logger log = LoggerFactory.getLogger(NotificationListener.class);
    
    @RabbitListener(queues = RabbitMQConfig.NOTIFICATION_QUEUE)
    public void handleNotification(String message) {
        log.info("Received notification: {}", message);
        // Traiter la notification
    }
}

4.4 Multiple Listeners (Scaling)

@Component
public class OrderEventListener {
    
    // 5 consumers en parallèle
    @RabbitListener(queues = RabbitMQConfig.ORDER_QUEUE, concurrency = "5")
    public void handleOrderEvent(OrderEvent event) {
        log.info("Processing on thread: {}", Thread.currentThread().getName());
        processOrder(event);
    }
}

---

5. Dead Letter Queue (DLQ)

5.1 Configuration DLQ

@Configuration
public class DeadLetterConfig {
    
    public static final String DLX_EXCHANGE = "dlx.exchange";
    public static final String DLQ_ORDER = "dlq.order";
    
    @Bean
    public DirectExchange deadLetterExchange() {
        return new DirectExchange(DLX_EXCHANGE);
    }
    
    @Bean
    public Queue deadLetterQueue() {
        return QueueBuilder.durable(DLQ_ORDER).build();
    }
    
    @Bean
    public Binding deadLetterBinding() {
        return BindingBuilder
            .bind(deadLetterQueue())
            .to(deadLetterExchange())
            .with("dlx.order");
    }
}

5.2 Listener DLQ

@Component
public class DeadLetterListener {
    
    private static final Logger log = LoggerFactory.getLogger(DeadLetterListener.class);
    
    private final OrderRepository orderRepository;
    private final NotificationService notificationService;
    
    @RabbitListener(queues = "dlq.order")
    public void handleDeadLetter(OrderEvent event, 
            @Header("x-death") List<Map<String, Object>> xDeath) {
        
        log.error("Dead letter received: {}", event);
        
        if (xDeath != null && !xDeath.isEmpty()) {
            Map<String, Object> death = xDeath.get(0);
            log.error("Death reason: {}, count: {}", 
                death.get("reason"), death.get("count"));
        }
        
        // Actions possibles:
        // 1. Sauvegarder pour analyse
        // 2. Notifier l'équipe ops
        // 3. Retry manuel
        
        notificationService.alertOpsTeam(
            "Order processing failed: " + event.getOrderId()
        );
    }
}

---

6. Topic Exchange (Routing Avancé)

6.1 Configuration

@Configuration
public class TopicExchangeConfig {
    
    public static final String TOPIC_EXCHANGE = "events.topic";
    
    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange(TOPIC_EXCHANGE);
    }
    
    // Queue pour tous les événements de commande
    @Bean
    public Queue allOrdersQueue() {
        return new Queue("queue.orders.all");
    }
    
    // Queue pour les commandes créées uniquement
    @Bean
    public Queue createdOrdersQueue() {
        return new Queue("queue.orders.created");
    }
    
    // Queue pour tous les événements
    @Bean
    public Queue allEventsQueue() {
        return new Queue("queue.events.all");
    }
    
    @Bean
    public Binding allOrdersBinding(Queue allOrdersQueue, TopicExchange topicExchange) {
        return BindingBuilder
            .bind(allOrdersQueue)
            .to(topicExchange)
            .with("order.*");  // order.created, order.shipped, etc.
    }
    
    @Bean
    public Binding createdOrdersBinding(Queue createdOrdersQueue, 
            TopicExchange topicExchange) {
        return BindingBuilder
            .bind(createdOrdersQueue)
            .to(topicExchange)
            .with("order.created");
    }
    
    @Bean
    public Binding allEventsBinding(Queue allEventsQueue, TopicExchange topicExchange) {
        return BindingBuilder
            .bind(allEventsQueue)
            .to(topicExchange)
            .with("#");  // Tous les messages
    }
}

6.2 Utilisation

@Service
public class EventPublisher {
    
    private final RabbitTemplate rabbitTemplate;
    
    public void publishOrderCreated(OrderEvent event) {
        rabbitTemplate.convertAndSend(
            "events.topic", 
            "order.created",  // Routing key
            event
        );
    }
    
    public void publishOrderShipped(OrderEvent event) {
        rabbitTemplate.convertAndSend(
            "events.topic", 
            "order.shipped", 
            event
        );
    }
    
    public void publishPaymentReceived(PaymentEvent event) {
        rabbitTemplate.convertAndSend(
            "events.topic", 
            "payment.received", 
            event
        );
    }
}

---

7. Request-Reply Pattern

7.1 Configuration

@Configuration
public class RpcConfig {
    
    public static final String RPC_QUEUE = "rpc.queue";
    public static final String RPC_EXCHANGE = "rpc.exchange";
    
    @Bean
    public Queue rpcQueue() {
        return new Queue(RPC_QUEUE);
    }
    
    @Bean
    public DirectExchange rpcExchange() {
        return new DirectExchange(RPC_EXCHANGE);
    }
    
    @Bean
    public Binding rpcBinding() {
        return BindingBuilder
            .bind(rpcQueue())
            .to(rpcExchange())
            .with("rpc");
    }
}

7.2 Client RPC

@Service
public class RpcClient {
    
    private final RabbitTemplate rabbitTemplate;
    
    public RpcClient(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }
    
    public String callService(String request) {
        Object response = rabbitTemplate.convertSendAndReceive(
            RpcConfig.RPC_EXCHANGE,
            "rpc",
            request
        );
        
        return response != null ? response.toString() : null;
    }
}

7.3 Serveur RPC

@Component
public class RpcServer {
    
    @RabbitListener(queues = RpcConfig.RPC_QUEUE)
    public String handleRequest(String request) {
        log.info("Received RPC request: {}", request);
        
        // Traitement
        String result = processRequest(request);
        
        log.info("Sending RPC response: {}", result);
        return result;
    }
    
    private String processRequest(String request) {
        // Logique métier
        return "Processed: " + request;
    }
}

---

8. Résumé

Patterns de Messaging

Pattern	Exchange	Usage
Point-to-point	Direct	Un consommateur
Publish/Subscribe	Fanout	Broadcast
Routing	Topic	Routage par pattern
Request/Reply	Direct	RPC

Bonnes Pratiques

1. Utiliser des messages idempotents
2. Implémenter une Dead Letter Queue
3. Configurer les retry appropriés
4. Monitorer les queues (RabbitMQ Management)
5. Utiliser la sérialisation JSON
6. Documenter les formats de messages
7. Tester avec des messages invalides

---

Chapitre Suivant: Tâches Planifiées

Retour au Module 14