Capítulo 3: Messaging Assíncrono com RabbitMQ

Objetivos do Capítulo

• Compreender os conceitos de messaging assíncrono
• Configurar RabbitMQ com Spring Boot
• Implementar produtores e consumidores
• Gerenciar exchanges, queues e bindings
• Implementar padrões de messaging

---

1. Introdução ao Messaging

1.1 Por que Messaging Assíncrono

Síncrono (REST):
Service A --HTTP--> Service B --HTTP--> Service C
    └── Aguarda resposta ──┘     └── Aguarda resposta ──┘
    
Problemas:
- Acoplamento temporal
- Service B fora = Service A falha
- Latência acumulada

Assíncrono (Messaging):
Service A --publish--> [Queue] --consume--> Service B
    └── Continua imediatamente

Vantagens:
- Desacoplamento
- Resiliência
- Escalabilidade
- Gerenciamento de picos

1.2 Conceitos do RabbitMQ

Conceito	Descrição
Producer	Envia mensagens
Consumer	Recebe mensagens
Queue	Fila de mensagens
Exchange	Roteia mensagens para as queues
Binding	Regra de roteamento exchange -> queue
Routing Key	Chave para roteamento

1.3 Tipos de Exchange

Direct Exchange:
Producer --> [Exchange] --> Queue (routing_key correspondência exata)

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

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

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

---

2. Configuração Spring AMQP

2.1 Dependências Maven

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

2.2 Configuração 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 Configuração 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 {
    
    // Nomes das queues e 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 para pedidos
    @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 pedidos
    @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 para notificações
    @Bean
    public Queue notificationQueue() {
        return QueueBuilder.durable(NOTIFICATION_QUEUE).build();
    }
    
    // Exchange Fanout para notificações (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);
    }
    
    // Conversor 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. Producer (Envio de Mensagens)

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 e 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("Enviando evento de pedido criado: {}", event);
        
        rabbitTemplate.convertAndSend(
            RabbitMQConfig.ORDER_EXCHANGE,
            RabbitMQConfig.ORDER_ROUTING_KEY,
            event
        );
        
        log.info("Evento de pedido enviado com sucesso");
    }
    
    public void sendNotification(String message) {
        log.info("Broadcasting notification: {}", message);
        
        // Fanout - não precisa de routing key
        rabbitTemplate.convertAndSend(
            RabbitMQConfig.NOTIFICATION_EXCHANGE,
            "",
            message
        );
    }
    
    // Envio com confirmação
    public void sendWithConfirmation(OrderEvent event) {
        rabbitTemplate.invoke(operations -> {
            operations.convertAndSend(
                RabbitMQConfig.ORDER_EXCHANGE,
                RabbitMQConfig.ORDER_ROUTING_KEY,
                event
            );
            
            // Aguarda a confirmação
            boolean confirmed = operations.waitForConfirms(5000);
            if (!confirmed) {
                throw new RuntimeException("Mensagem não confirmada pelo broker");
            }
            
            return null;
        });
    }
}

3.3 Uso em um 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. Criar o pedido
        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 o evento
        OrderEvent event = new OrderEvent(
            saved.getId(),
            saved.getUserId(),
            saved.getProductName(),
            saved.getTotalAmount(),
            saved.getStatus()
        );
        
        eventProducer.sendOrderCreatedEvent(event);
        
        return saved;
    }
}

---

4. Consumer (Recepção de Mensagens)

4.1 Listener Simples

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("Recebeu evento de pedido: {}", event);
        
        try {
            // Processamento da mensagem
            processOrder(event);
            log.info("Evento de pedido processado com sucesso");
        } catch (Exception e) {
            log.error("Erro ao processar evento de pedido: {}", e.getMessage());
            throw e; // Re-throw para retry/DLQ
        }
    }
    
    private void processOrder(OrderEvent event) {
        // Lógica de negócio
        // - Enviar e-mail de confirmação
        // - Atualizar estoque
        // - Notificar sistema de entrega
        log.info("Processando pedido {} para usuário {}", 
            event.getOrderId(), event.getUserId());
    }
}

4.2 Listener com Confirmação 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("Recebeu evento de pedido: {}", event);
        
        try {
            processOrder(event);
            
            // Acknowledge sucesso
            channel.basicAck(deliveryTag, false);
            log.info("Mensagem reconhecida");
            
        } catch (Exception e) {
            log.error("Erro ao processar: {}", e.getMessage());
            
            try {
                // Rejeita e requeue (ou envia para DLQ)
                channel.basicNack(deliveryTag, false, false);
            } catch (IOException ioEx) {
                log.error("Erro ao rejeitar mensagem", ioEx);
            }
        }
    }
}

4.3 Listener para Notificações (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("Recebeu notificação: {}", message);
        // Processar a notificação
    }
}

4.4 Múltiplos Listeners (Scaling)

@Component
public class OrderEventListener {
    
    // 5 consumers em paralelo
    @RabbitListener(queues = RabbitMQConfig.ORDER_QUEUE, concurrency = "5")
    public void handleOrderEvent(OrderEvent event) {
        log.info("Processando na thread: {}", Thread.currentThread().getName());
        processOrder(event);
    }
}

---

5. Dead Letter Queue (DLQ)

5.1 Configuração 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 recebida: {}", event);
        
        if (xDeath != null && !xDeath.isEmpty()) {
            Map<String, Object> death = xDeath.get(0);
            log.error("Motivo da morte: {}, contagem: {}", 
                death.get("reason"), death.get("count"));
        }
        
        // Ações possíveis:
        // 1. Salvar para análise
        // 2. Notificar equipe de operações
        // 3. Retry manual
        
        notificationService.alertOpsTeam(
            "Falha no processamento do pedido: " + event.getOrderId()
        );
    }
}

---

6. Topic Exchange (Roteamento Avançado)

6.1 Configuração

@Configuration
public class TopicExchangeConfig {
    
    public static final String TOPIC_EXCHANGE = "events.topic";
    
    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange(TOPIC_EXCHANGE);
    }
    
    // Queue para todos os eventos de pedido
    @Bean
    public Queue allOrdersQueue() {
        return new Queue("queue.orders.all");
    }
    
    // Queue apenas para pedidos criados
    @Bean
    public Queue createdOrdersQueue() {
        return new Queue("queue.orders.created");
    }
    
    // Queue para todos os 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("#");  // Todas as mensagens
    }
}

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 Configuração

@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("Recebeu requisição RPC: {}", request);
        
        // Processamento
        String result = processRequest(request);
        
        log.info("Enviando resposta RPC: {}", result);
        return result;
    }
    
    private String processRequest(String request) {
        // Lógica de negócio
        return "Processed: " + request;
    }
}

---

8. Resumo

Padrões de Messaging

Padrão	Exchange	Uso
Point-to-point	Direct	Um consumidor
Publish/Subscribe	Fanout	Broadcast
Routing	Topic	Roteamento por padrão
Request/Reply	Direct	RPC

Boas Práticas

1. Utilizar mensagens idempotentes
2. Implementar uma Dead Letter Queue
3. Configurar retries apropriados
4. Monitorar as queues (RabbitMQ Management)
5. Utilizar serialização JSON
6. Documentar os formatos de mensagens
7. Testar com mensagens inválidas

---

Próximo Capítulo: Tarefas Agendadas

Voltar ao Módulo 14