Capítulo 3: Mensajería Asíncrona con RabbitMQ

Objetivos del Capítulo

• Comprender los conceptos de mensajería asíncrona
• Configurar RabbitMQ con Spring Boot
• Implementar productores y consumidores
• Gestionar exchanges, colas y bindings
• Implementar patrones de mensajería

---

1. Introducción a la Mensajería

1.1 Por qué la Mensajería Asíncrona

Sincrónico (REST):
Service A --HTTP--> Service B --HTTP--> Service C
    └── Espera respuesta ──┘     └── Espera respuesta ──┘
    
Problemas:
- Acoplamiento temporal
- Service B caído = Service A falla
- Latencia acumulada

Asíncrono (Messaging):
Service A --publish--> [Queue] --consume--> Service B
    └── Continúa inmediatamente

Ventajas:
- Desacoplamiento
- Resiliencia
- Escalabilidad
- Manejo de picos

1.2 Conceptos de RabbitMQ

Concepto	Descripción
Producer	Envía mensajes
Consumer	Recibe mensajes
Queue	Cola de mensajes
Exchange	Enruta los mensajes a las colas
Binding	Regla de enrutamiento exchange -> queue
Routing Key	Clave para el enrutamiento

1.3 Tipos de 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. Configuración de Spring AMQP

2.1 Dependencias Maven

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

2.2 Configuración 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 Configuración 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 {
    
    // Nombres de las colas y 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";
    
    // Cola para las órdenes
    @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 para las órdenes
    @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);
    }
    
    // Cola para notificaciones
    @Bean
    public Queue notificationQueue() {
        return QueueBuilder.durable(NOTIFICATION_QUEUE).build();
    }
    
    // Exchange Fanout para notificaciones (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);
    }
    
    // Convertidor JSON
    @Bean
    public MessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }
    
    // Template configurado
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        template.setMessageConverter(jsonMessageConverter());
        return template;
    }
}

---

3. Productor (Envío de Mensajes)

3.1 DTO de Mensaje

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 y 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 Servicio Productor

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("Enviando evento de orden creada: {}", event);
        
        rabbitTemplate.convertAndSend(
            RabbitMQConfig.ORDER_EXCHANGE,
            RabbitMQConfig.ORDER_ROUTING_KEY,
            event
        );
        
        log.info("Evento de orden enviado exitosamente");
    }
    
    public void sendNotification(String message) {
        log.info("Transmitiendo notificación: {}", message);
        
        // Fanout - no se necesita routing key
        rabbitTemplate.convertAndSend(
            RabbitMQConfig.NOTIFICATION_EXCHANGE,
            "",
            message
        );
    }
    
    // Envío con confirmación
    public void sendWithConfirmation(OrderEvent event) {
        rabbitTemplate.invoke(operations -> {
            operations.convertAndSend(
                RabbitMQConfig.ORDER_EXCHANGE,
                RabbitMQConfig.ORDER_ROUTING_KEY,
                event
            );
            
            // Esperar la confirmación
            boolean confirmed = operations.waitForConfirms(5000);
            if (!confirmed) {
                throw new RuntimeException("Mensaje no confirmado por el broker");
            }
            
            return null;
        });
    }
}

3.3 Uso en un Servicio

@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. Crear la orden
        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. Publicar el evento
        OrderEvent event = new OrderEvent(
            saved.getId(),
            saved.getUserId(),
            saved.getProductName(),
            saved.getTotalAmount(),
            saved.getStatus()
        );
        
        eventProducer.sendOrderCreatedEvent(event);
        
        return saved;
    }
}

---

4. Consumidor (Recepción de Mensajes)

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("Evento de orden recibido: {}", event);
        
        try {
            // Procesamiento del mensaje
            processOrder(event);
            log.info("Evento de orden procesado exitosamente");
        } catch (Exception e) {
            log.error("Error al procesar el evento de orden: {}", e.getMessage());
            throw e; // Re-lanzar para retry/DLQ
        }
    }
    
    private void processOrder(OrderEvent event) {
        // Lógica de negocio
        // - Enviar correo de confirmación
        // - Actualizar inventario
        // - Notificar al sistema de entrega
        log.info("Procesando orden {} para usuario {}", 
            event.getOrderId(), event.getUserId());
    }
}

4.2 Listener con Acuse de Recibo Manual

@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("Evento de orden recibido: {}", event);
        
        try {
            processOrder(event);
            
            // Confirmar éxito
            channel.basicAck(deliveryTag, false);
            log.info("Mensaje confirmado");
            
        } catch (Exception e) {
            log.error("Error al procesar: {}", e.getMessage());
            
            try {
                // Rechazar y reencolar (o enviar a DLQ)
                channel.basicNack(deliveryTag, false, false);
            } catch (IOException ioEx) {
                log.error("Error al rechazar el mensaje", ioEx);
            }
        }
    }
}

4.3 Listener para Notificaciones (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("Notificación recibida: {}", message);
        // Procesar la notificación
    }
}

4.4 Múltiples Listeners (Escalado)

@Component
public class OrderEventListener {
    
    // 5 consumidores en paralelo
    @RabbitListener(queues = RabbitMQConfig.ORDER_QUEUE, concurrency = "5")
    public void handleOrderEvent(OrderEvent event) {
        log.info("Procesando en el hilo: {}", Thread.currentThread().getName());
        processOrder(event);
    }
}

---

5. Dead Letter Queue (DLQ)

5.1 Configuración 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 recibida: {}", event);
        
        if (xDeath != null && !xDeath.isEmpty()) {
            Map<String, Object> death = xDeath.get(0);
            log.error("Razón de muerte: {}, conteo: {}", 
                death.get("reason"), death.get("count"));
        }
        
        // Acciones posibles:
        // 1. Guardar para análisis
        // 2. Notificar al equipo de operaciones
        // 3. Reintento manual
        
        notificationService.alertOpsTeam(
            "Fallo en el procesamiento de la orden: " + event.getOrderId()
        );
    }
}

---

6. Topic Exchange (Enrutamiento Avanzado)

6.1 Configuración

@Configuration
public class TopicExchangeConfig {
    
    public static final String TOPIC_EXCHANGE = "events.topic";
    
    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange(TOPIC_EXCHANGE);
    }
    
    // Cola para todos los eventos de orden
    @Bean
    public Queue allOrdersQueue() {
        return new Queue("queue.orders.all");
    }
    
    // Cola solo para órdenes creadas
    @Bean
    public Queue createdOrdersQueue() {
        return new Queue("queue.orders.created");
    }
    
    // Cola para todos los eventos
    @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("#");  // Todos los mensajes
    }
}

6.2 Uso

@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 Configuración

@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 Cliente 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 Servidor RPC

@Component
public class RpcServer {
    
    @RabbitListener(queues = RpcConfig.RPC_QUEUE)
    public String handleRequest(String request) {
        log.info("Solicitud RPC recibida: {}", request);
        
        // Procesamiento
        String result = processRequest(request);
        
        log.info("Enviando respuesta RPC: {}", result);
        return result;
    }
    
    private String processRequest(String request) {
        // Lógica de negocio
        return "Procesado: " + request;
    }
}

---

8. Resumen

Patrones de Mensajería

Patrón	Exchange	Uso
Punto a punto	Direct	Un consumidor
Publicar/Suscribir	Fanout	Broadcast
Enrutamiento	Topic	Enrutamiento por patrón
Request/Reply	Direct	RPC

Buenas Prácticas

1. Utilizar mensajes idempotentes
2. Implementar una Dead Letter Queue
3. Configurar los reintentos apropiados
4. Monitorear las colas (RabbitMQ Management)
5. Utilizar serialización JSON
6. Documentar los formatos de mensajes
7. Probar con mensajes inválidos

---

Capítulo Siguiente: Tareas Programadas

Volver al Módulo 14