RabbitMQ

// Các properties quan trọng:
MessageProperties properties = MessagePropertiesBuilder.newInstance()
    .setContentType("application/json")       // MIME type của body
    .setContentEncoding("UTF-8")              // Encoding
    .setDeliveryMode(MessageDeliveryMode.PERSISTENT) // 1=transient, 2=persistent
    .setPriority(5)                           // 0-9, priority queue
    .setCorrelationId("abc-123")              // Dùng cho RPC pattern
    .setReplyTo("response.queue")             // Queue để gửi response
    .setExpiration("60000")                   // TTL tính bằng milliseconds
    .setMessageId(UUID.randomUUID().toString()) // Unique ID
    .setTimestamp(new Date())                 // Timestamp
    .setType("OrderCreatedEvent")             // Message type
    .setAppId("order-service")                // Application ID
    .build();

// Ba loại acknowledgment:
channel.basicAck(deliveryTag, false);   // Xử lý thành công, xóa message
channel.basicNack(deliveryTag, false, true);  // Thất bại, requeue
channel.basicReject(deliveryTag, false);      // Thất bại, không requeue (→ DLX)

// Khai báo
channel.exchangeDeclare("order.direct", BuiltinExchangeType.DIRECT, true);
channel.queueDeclare("order-processing", true, false, false, null);
channel.queueBind("order-processing", "order.direct", "order.created");
 
// Publish
channel.basicPublish("order.direct", "order.created", null, message.getBytes());

// Khai báo
channel.exchangeDeclare("notification.fanout", BuiltinExchangeType.FANOUT, true);
channel.queueBind("email-queue", "notification.fanout", ""); // routing key bị ignore
 
// Publish — routing key không quan trọng
channel.basicPublish("notification.fanout", "", null, message.getBytes());

// Khai báo
channel.exchangeDeclare("events.topic", BuiltinExchangeType.TOPIC, true);
channel.queueBind("vn-orders", "events.topic", "order.*.vn");
channel.queueBind("all-orders", "events.topic", "order.#");
channel.queueBind("all-events", "events.topic", "#");
 
// Publish
channel.basicPublish("events.topic", "order.created.vn", null, message.getBytes());

// Binding với headers
Map<String, Object> headers = new HashMap<>();
headers.put("x-match", "all");  // "all" = AND, "any" = OR
headers.put("format", "pdf");
headers.put("type", "report");
channel.queueBind("pdf-reports", "docs.headers", "", headers);
 
// Publish với matching headers
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
    .headers(Map.of("format", "pdf", "type", "report"))
    .build();
channel.basicPublish("docs.headers", "", props, message.getBytes());

<dependencies>
    <!-- Spring Boot Starter -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
 
    <!-- Spring AMQP — RabbitMQ integration -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
 
    <!-- JSON serialization -->
    <dependency>
        <groupId>com.fasterxml.jackson.core</groupId>
        <artifactId>jackson-databind</artifactId>
    </dependency>
 
    <!-- Lombok — giảm boilerplate -->
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <optional>true</optional>
    </dependency>
</dependencies>

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtual-host: /
    # Connection pooling
    connection-timeout: 5000
    # Listener configuration
    listener:
      simple:
        acknowledge-mode: manual # QUAN TRỌNG: Manual ack cho production
        prefetch: 10 # Số message consumer nhận trước khi ack
        concurrency: 3 # Min consumer threads
        max-concurrency: 10 # Max consumer threads
        retry:
          enabled: true
          initial-interval: 1000 # 1 giây
          max-interval: 10000 # 10 giây
          multiplier: 2.0 # Exponential backoff
          max-attempts: 3
    # Publisher confirms
    publisher-confirm-type: correlated # Async publisher confirms
    publisher-returns: true # Return unroutable messages

package com.example.order.config;
 
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
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 {
 
    // ==================== Constants ====================
    public static final String ORDER_EXCHANGE = "order.exchange";
    public static final String ORDER_QUEUE = "order.processing.queue";
    public static final String ORDER_ROUTING_KEY = "order.created";
 
    public static final String NOTIFICATION_EXCHANGE = "notification.exchange";
    public static final String EMAIL_QUEUE = "notification.email.queue";
    public static final String SMS_QUEUE = "notification.sms.queue";
 
    // Dead Letter
    public static final String DLX_EXCHANGE = "dlx.exchange";
    public static final String DLX_QUEUE = "dlx.queue";
    public static final String DLX_ROUTING_KEY = "dlx.routing";
 
    // ==================== Exchanges ====================
 
    @Bean
    public TopicExchange orderExchange() {
        return ExchangeBuilder
                .topicExchange(ORDER_EXCHANGE)
                .durable(true)
                .build();
    }
 
    @Bean
    public FanoutExchange notificationExchange() {
        return ExchangeBuilder
                .fanoutExchange(NOTIFICATION_EXCHANGE)
                .durable(true)
                .build();
    }
 
    @Bean
    public DirectExchange deadLetterExchange() {
        return ExchangeBuilder
                .directExchange(DLX_EXCHANGE)
                .durable(true)
                .build();
    }
 
    // ==================== Queues ====================
 
    @Bean
    public Queue orderQueue() {
        return QueueBuilder
                .durable(ORDER_QUEUE)
                .withArgument("x-dead-letter-exchange", DLX_EXCHANGE)
                .withArgument("x-dead-letter-routing-key", DLX_ROUTING_KEY)
                .withArgument("x-message-ttl", 300000) // 5 phút TTL
                .build();
    }
 
    @Bean
    public Queue emailQueue() {
        return QueueBuilder
                .durable(EMAIL_QUEUE)
                .withArgument("x-dead-letter-exchange", DLX_EXCHANGE)
                .withArgument("x-dead-letter-routing-key", DLX_ROUTING_KEY)
                .build();
    }
 
    @Bean
    public Queue smsQueue() {
        return QueueBuilder
                .durable(SMS_QUEUE)
                .withArgument("x-dead-letter-exchange", DLX_EXCHANGE)
                .withArgument("x-dead-letter-routing-key", DLX_ROUTING_KEY)
                .build();
    }
 
    @Bean
    public Queue deadLetterQueue() {
        return QueueBuilder
                .durable(DLX_QUEUE)
                .build();
    }
 
    // ==================== Bindings ====================
 
    @Bean
    public Binding orderBinding() {
        return BindingBuilder
                .bind(orderQueue())
                .to(orderExchange())
                .with("order.#"); // Match tất cả order events
    }
 
    @Bean
    public Binding emailBinding() {
        return BindingBuilder
                .bind(emailQueue())
                .to(notificationExchange());
    }
 
    @Bean
    public Binding smsBinding() {
        return BindingBuilder
                .bind(smsQueue())
                .to(notificationExchange());
    }
 
    @Bean
    public Binding deadLetterBinding() {
        return BindingBuilder
                .bind(deadLetterQueue())
                .to(deadLetterExchange())
                .with(DLX_ROUTING_KEY);
    }
 
    // ==================== Message Converter ====================
 
    @Bean
    public MessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }
 
    // ==================== RabbitTemplate ====================
 
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        template.setMessageConverter(jsonMessageConverter());
 
        // Publisher Confirms — biết chắc message đã đến Exchange
        template.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                log.debug("Message delivered to exchange: {}", correlationData);
            } else {
                log.error("Message FAILED to deliver to exchange: {}, cause: {}",
                        correlationData, cause);
                // TODO: Retry logic hoặc save to DB
            }
        });
 
        // Publisher Returns — message không route được đến queue nào
        template.setReturnsCallback(returned -> {
            log.error("Message returned: exchange={}, routingKey={}, replyCode={}, replyText={}",
                    returned.getExchange(),
                    returned.getRoutingKey(),
                    returned.getReplyCode(),
                    returned.getReplyText());
        });
 
        template.setMandatory(true); // Enable returns cho unroutable messages
        return template;
    }
 
    // ==================== Listener Container Factory ====================
 
    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
            ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(jsonMessageConverter());
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        factory.setPrefetchCount(10);
        factory.setConcurrentConsumers(3);
        factory.setMaxConcurrentConsumers(10);
        return factory;
    }
}

package com.example.order.model;
 
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.io.Serializable;
import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.util.List;
 
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class OrderEvent implements Serializable {
 
    private String orderId;
    private String customerId;
    private String customerEmail;
    private String customerPhone;
    private List<OrderItem> items;
    private BigDecimal totalAmount;
    private String currency;
    private OrderStatus status;
    private LocalDateTime createdAt;
    private String eventType; // "created", "paid", "shipped", "cancelled"
 
    @Data
    @Builder
    @NoArgsConstructor
    @AllArgsConstructor
    public static class OrderItem implements Serializable {
        private String productId;
        private String productName;
        private int quantity;
        private BigDecimal price;
    }
 
    public enum OrderStatus {
        PENDING, CONFIRMED, PAID, SHIPPED, DELIVERED, CANCELLED
    }
}

package com.example.order.producer;
 
import com.example.order.config.RabbitMQConfig;
import com.example.order.model.OrderEvent;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;
 
import java.util.UUID;
 
@Slf4j
@Service
@RequiredArgsConstructor
public class OrderProducer {
 
    private final RabbitTemplate rabbitTemplate;
 
    /**
     * Publish order event đến Topic Exchange.
     * Routing key format: order.{eventType}
     * Ví dụ: order.created, order.paid, order.shipped
     */
    public void publishOrderEvent(OrderEvent event) {
        String routingKey = "order." + event.getEventType();
        String correlationId = UUID.randomUUID().toString();
 
        log.info("Publishing order event: orderId={}, type={}, correlationId={}",
                event.getOrderId(), event.getEventType(), correlationId);
 
        CorrelationData correlationData = new CorrelationData(correlationId);
 
        rabbitTemplate.convertAndSend(
                RabbitMQConfig.ORDER_EXCHANGE,
                routingKey,
                event,
                message -> {
                    // Custom message properties
                    message.getMessageProperties().setMessageId(UUID.randomUUID().toString());
                    message.getMessageProperties().setCorrelationId(correlationId);
                    message.getMessageProperties().setHeader("source", "order-service");
                    message.getMessageProperties().setHeader("version", "1.0");
                    return message;
                },
                correlationData
        );
 
        log.info("Order event published successfully: orderId={}", event.getOrderId());
    }
 
    /**
     * Broadcast notification đến tất cả notification consumers.
     * Dùng Fanout Exchange — tất cả queue bound sẽ nhận message.
     */
    public void broadcastNotification(OrderEvent event) {
        log.info("Broadcasting notification for order: {}", event.getOrderId());
 
        rabbitTemplate.convertAndSend(
                RabbitMQConfig.NOTIFICATION_EXCHANGE,
                "", // Fanout exchange bỏ qua routing key
                event
        );
    }
}

package com.example.order.consumer;
 
import com.example.order.model.OrderEvent;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;
 
import java.io.IOException;
 
@Slf4j
@Component
public class OrderConsumer {
 
    /**
     * Consumer chính xử lý order events.
     *
     * Manual Ack pattern:
     * 1. Nhận message
     * 2. Xử lý business logic
     * 3. Ack nếu thành công, Nack nếu thất bại
     *
     * QUAN TRỌNG: Luôn ack/nack trong finally block để tránh message bị stuck
     */
    @RabbitListener(
            queues = "order.processing.queue",
            containerFactory = "rabbitListenerContainerFactory"
    )
    public void handleOrderEvent(
            @Payload OrderEvent event,
            @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag,
            @Header(AmqpHeaders.REDELIVERED) boolean redelivered,
            Channel channel) throws IOException {
 
        log.info("Received order event: orderId={}, type={}, redelivered={}",
                event.getOrderId(), event.getEventType(), redelivered);
 
        try {
            // ========== Business Logic ==========
            switch (event.getEventType()) {
                case "created":
                    processNewOrder(event);
                    break;
                case "paid":
                    processPayment(event);
                    break;
                case "shipped":
                    processShipment(event);
                    break;
                case "cancelled":
                    processCancellation(event);
                    break;
                default:
                    log.warn("Unknown event type: {}", event.getEventType());
            }
 
            // ========== ACK — Xử lý thành công ==========
            channel.basicAck(deliveryTag, false);
            log.info("Order event processed successfully: orderId={}", event.getOrderId());
 
        } catch (BusinessException e) {
            // Business error — không cần retry, gửi đến DLX
            log.error("Business error processing order: {}", event.getOrderId(), e);
            channel.basicReject(deliveryTag, false); // false = không requeue → đến DLX
 
        } catch (TransientException e) {
            // Transient error (DB timeout, network issue) — có thể retry
            log.warn("Transient error processing order: {}, redelivered: {}",
                    event.getOrderId(), redelivered);
 
            if (redelivered) {
                // Đã retry rồi mà vẫn fail → gửi đến DLX
                channel.basicReject(deliveryTag, false);
                log.error("Message sent to DLX after retry: orderId={}", event.getOrderId());
            } else {
                // Retry lần đầu — requeue
                channel.basicNack(deliveryTag, false, true); // true = requeue
            }
 
        } catch (Exception e) {
            // Unexpected error
            log.error("Unexpected error processing order: {}", event.getOrderId(), e);
            channel.basicReject(deliveryTag, false); // Gửi đến DLX
        }
    }
 
    private void processNewOrder(OrderEvent event) {
        log.info("Processing new order: {} - Amount: {} {}",
                event.getOrderId(), event.getTotalAmount(), event.getCurrency());
        // Validate inventory, reserve stock, create order record...
    }
 
    private void processPayment(OrderEvent event) {
        log.info("Processing payment for order: {}", event.getOrderId());
        // Update order status, trigger fulfillment...
    }
 
    private void processShipment(OrderEvent event) {
        log.info("Processing shipment for order: {}", event.getOrderId());
        // Update tracking info, notify customer...
    }
 
    private void processCancellation(OrderEvent event) {
        log.info("Processing cancellation for order: {}", event.getOrderId());
        // Release inventory, process refund...
    }
}

package com.example.order.consumer;
 
import com.example.order.model.OrderEvent;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;
 
import java.io.IOException;
 
@Slf4j
@Component
public class NotificationConsumer {
 
    /**
     * Email notification consumer.
     * Nhận từ Fanout Exchange — mọi order event đều trigger email.
     */
    @RabbitListener(queues = "notification.email.queue")
    public void handleEmailNotification(
            @Payload OrderEvent event,
            @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag,
            Channel channel) throws IOException {
 
        try {
            log.info("Sending email notification for order: {} to {}",
                    event.getOrderId(), event.getCustomerEmail());
 
            // Gửi email logic...
            // emailService.sendOrderConfirmation(event);
 
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            log.error("Failed to send email for order: {}", event.getOrderId(), e);
            channel.basicNack(deliveryTag, false, true); // Retry
        }
    }
 
    /**
     * SMS notification consumer.
     * Cũng nhận từ Fanout Exchange — cùng message, khác channel.
     */
    @RabbitListener(queues = "notification.sms.queue")
    public void handleSmsNotification(
            @Payload OrderEvent event,
            @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag,
            Channel channel) throws IOException {
 
        try {
            log.info("Sending SMS notification for order: {} to {}",
                    event.getOrderId(), event.getCustomerPhone());
 
            // Gửi SMS logic...
            // smsService.sendOrderSms(event);
 
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            log.error("Failed to send SMS for order: {}", event.getOrderId(), e);
            channel.basicNack(deliveryTag, false, true);
        }
    }
}

package com.example.order.controller;
 
import com.example.order.model.OrderEvent;
import com.example.order.producer.OrderProducer;
import lombok.RequiredArgsConstructor;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
 
import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.UUID;
 
@RestController
@RequestMapping("/api/orders")
@RequiredArgsConstructor
public class OrderController {
 
    private final OrderProducer orderProducer;
 
    @PostMapping
    public ResponseEntity<Map<String, String>> createOrder(@RequestBody OrderEvent orderRequest) {
        // Generate order ID
        String orderId = "ORD-" + UUID.randomUUID().toString().substring(0, 8).toUpperCase();
        orderRequest.setOrderId(orderId);
        orderRequest.setStatus(OrderEvent.OrderStatus.PENDING);
        orderRequest.setCreatedAt(LocalDateTime.now());
        orderRequest.setEventType("created");
 
        // Publish to order processing queue
        orderProducer.publishOrderEvent(orderRequest);
 
        // Broadcast notification
        orderProducer.broadcastNotification(orderRequest);
 
        // Response ngay lập tức — không chờ processing
        return ResponseEntity.accepted().body(Map.of(
                "orderId", orderId,
                "status", "ACCEPTED",
                "message", "Order is being processed"
        ));
    }
 
    /**
     * Demo endpoint để test các loại event
     */
    @PostMapping("/{orderId}/events/{eventType}")
    public ResponseEntity<Map<String, String>> publishEvent(
            @PathVariable String orderId,
            @PathVariable String eventType) {
 
        OrderEvent event = OrderEvent.builder()
                .orderId(orderId)
                .eventType(eventType)
                .createdAt(LocalDateTime.now())
                .build();
 
        orderProducer.publishOrderEvent(event);
 
        return ResponseEntity.accepted().body(Map.of(
                "orderId", orderId,
                "eventType", eventType,
                "status", "EVENT_PUBLISHED"
        ));
    }
}

# 1. Start RabbitMQ bằng Docker
docker run -d --name rabbitmq \
  -p 5672:5672 \
  -p 15672:15672 \
  rabbitmq:3-management
 
# Management UI: http://localhost:15672 (guest/guest)
 
# 2. Start Spring Boot app
./mvnw spring-boot:run
 
# 3. Tạo order
curl -X POST http://localhost:8080/api/orders \
  -H "Content-Type: application/json" \
  -d '{
    "customerId": "CUST-001",
    "customerEmail": "user@example.com",
    "customerPhone": "+84123456789",
    "items": [
      {
        "productId": "PROD-001",
        "productName": "Mechanical Keyboard",
        "quantity": 1,
        "price": 150.00
      }
    ],
    "totalAmount": 150.00,
    "currency": "USD"
  }'
 
# Response (ngay lập tức, < 10ms):
# {
#   "orderId": "ORD-A1B2C3D4",
#   "status": "ACCEPTED",
#   "message": "Order is being processed"
# }
 
# 4. Publish event
curl -X POST http://localhost:8080/api/orders/ORD-A1B2C3D4/events/paid
 
# 5. Xem logs — bạn sẽ thấy:
# OrderConsumer: Received order event: orderId=ORD-A1B2C3D4, type=created
# NotificationConsumer: Sending email notification for order: ORD-A1B2C3D4
# NotificationConsumer: Sending SMS notification for order: ORD-A1B2C3D4

@Slf4j
@Service
@RequiredArgsConstructor
public class ReliableProducer {
 
    private final RabbitTemplate rabbitTemplate;
    private final OrderOutboxRepository outboxRepository; // Outbox pattern
 
    /**
     * Transactional Outbox Pattern:
     * 1. Save message vào DB (cùng transaction với business logic)
     * 2. Publish message đến RabbitMQ
     * 3. Nếu publish thành công → mark as sent
     * 4. Nếu publish thất bại → scheduler retry
     */
    @Transactional
    public void publishWithOutbox(OrderEvent event) {
        // Step 1: Save to outbox table
        OrderOutbox outbox = OrderOutbox.builder()
                .messageId(UUID.randomUUID().toString())
                .exchange(RabbitMQConfig.ORDER_EXCHANGE)
                .routingKey("order." + event.getEventType())
                .payload(objectMapper.writeValueAsString(event))
                .status(OutboxStatus.PENDING)
                .createdAt(LocalDateTime.now())
                .build();
        outboxRepository.save(outbox);
 
        // Step 2: Try to publish
        try {
            CorrelationData correlationData = new CorrelationData(outbox.getMessageId());
            rabbitTemplate.convertAndSend(
                    outbox.getExchange(),
                    outbox.getRoutingKey(),
                    event,
                    correlationData
            );
 
            // Step 3: Mark as sent (confirm callback sẽ verify)
            outbox.setStatus(OutboxStatus.SENT);
            outboxRepository.save(outbox);
        } catch (Exception e) {
            log.error("Failed to publish message: {}", outbox.getMessageId(), e);
            // Message vẫn ở PENDING → scheduler sẽ retry
        }
    }
 
    /**
     * Scheduler retry cho messages PENDING quá lâu.
     * Chạy mỗi 30 giây.
     */
    @Scheduled(fixedDelay = 30000)
    public void retryPendingMessages() {
        List<OrderOutbox> pendingMessages = outboxRepository
                .findByStatusAndCreatedAtBefore(
                        OutboxStatus.PENDING,
                        LocalDateTime.now().minusMinutes(1)
                );
 
        for (OrderOutbox outbox : pendingMessages) {
            try {
                rabbitTemplate.convertAndSend(
                        outbox.getExchange(),
                        outbox.getRoutingKey(),
                        objectMapper.readValue(outbox.getPayload(), OrderEvent.class)
                );
                outbox.setStatus(OutboxStatus.SENT);
                outbox.setRetryCount(outbox.getRetryCount() + 1);
            } catch (Exception e) {
                outbox.setRetryCount(outbox.getRetryCount() + 1);
                if (outbox.getRetryCount() >= 5) {
                    outbox.setStatus(OutboxStatus.FAILED);
                    log.error("Message permanently failed after 5 retries: {}",
                            outbox.getMessageId());
                }
            }
            outboxRepository.save(outbox);
        }
    }
}

@Slf4j
@Component
@RequiredArgsConstructor
public class IdempotentOrderConsumer {
 
    private final ProcessedMessageRepository processedMessageRepo;
    private final OrderService orderService;
 
    @RabbitListener(queues = "order.processing.queue")
    public void handleOrderEvent(
            @Payload OrderEvent event,
            @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag,
            @Header(value = AmqpHeaders.MESSAGE_ID, required = false) String messageId,
            Channel channel) throws IOException {
 
        // Idempotency check — đã xử lý message này chưa?
        if (messageId != null && processedMessageRepo.existsByMessageId(messageId)) {
            log.warn("Duplicate message detected, skipping: messageId={}", messageId);
            channel.basicAck(deliveryTag, false); // Ack để không nhận lại
            return;
        }
 
        try {
            // Process order
            orderService.processOrder(event);
 
            // Mark as processed
            if (messageId != null) {
                processedMessageRepo.save(new ProcessedMessage(
                        messageId,
                        event.getOrderId(),
                        LocalDateTime.now()
                ));
            }
 
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            log.error("Error processing order: {}", event.getOrderId(), e);
            channel.basicReject(deliveryTag, false);
        }
    }
}

@Slf4j
@Component
@RequiredArgsConstructor
public class DeadLetterConsumer {
 
    private final RabbitTemplate rabbitTemplate;
    private final AlertService alertService;
 
    /**
     * Consumer cho Dead Letter Queue.
     * Phân tích nguyên nhân fail và quyết định:
     * 1. Retry (gửi lại main queue)
     * 2. Alert (thông báo team)
     * 3. Archive (lưu lại để debug)
     */
    @RabbitListener(queues = "dlx.queue")
    public void handleDeadLetter(
            org.springframework.amqp.core.Message message,
            @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag,
            Channel channel) throws IOException {
 
        // Lấy thông tin về nguyên nhân dead letter
        Map<String, Object> headers = message.getMessageProperties().getHeaders();
        List<Map<String, Object>> xDeath = (List<Map<String, Object>>) headers.get("x-death");
 
        if (xDeath != null && !xDeath.isEmpty()) {
            Map<String, Object> deathInfo = xDeath.get(0);
            String reason = (String) deathInfo.get("reason");      // "rejected", "expired", "maxlen"
            String queue = (String) deathInfo.get("queue");         // Queue gốc
            Long count = (Long) deathInfo.get("count");             // Số lần dead letter
            String exchange = (String) deathInfo.get("exchange");   // Exchange gốc
 
            log.error("Dead letter received: reason={}, queue={}, count={}, exchange={}",
                    reason, queue, count, exchange);
 
            // Retry logic dựa trên số lần fail
            if (count != null && count < 3) {
                // Retry — gửi lại queue gốc sau delay
                log.info("Retrying message (attempt {}): queue={}", count + 1, queue);
 
                // Delay trước khi retry (exponential backoff)
                try {
                    Thread.sleep(Math.min(1000 * (long) Math.pow(2, count), 30000));
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
 
                rabbitTemplate.convertAndSend(exchange,
                        message.getMessageProperties().getReceivedRoutingKey(),
                        message);
            } else {
                // Đã retry quá nhiều — alert và archive
                log.error("Message permanently failed after {} attempts. Alerting team.", count);
                alertService.sendAlert(
                        "Dead Letter Alert",
                        String.format("Message failed %d times in queue %s. Manual intervention needed.",
                                count, queue)
                );
                // Archive to database for later analysis
                // deadLetterRepository.save(...)
            }
        }
 
        channel.basicAck(deliveryTag, false);
    }
}

@Configuration
public class RetryQueueConfig {
 
    // Retry queue với TTL — message tự động quay lại main queue sau delay
    @Bean
    public Queue retryQueue() {
        return QueueBuilder
                .durable("order.retry.queue")
                .withArgument("x-dead-letter-exchange", RabbitMQConfig.ORDER_EXCHANGE)
                .withArgument("x-dead-letter-routing-key", "order.created")
                .withArgument("x-message-ttl", 5000) // 5 giây delay
                .build();
    }
 
    // Binding retry queue đến DLX exchange
    @Bean
    public Binding retryBinding() {
        return BindingBuilder
                .bind(retryQueue())
                .to(new DirectExchange(RabbitMQConfig.DLX_EXCHANGE))
                .with("retry.order");
    }
}

# application.yml
spring:
  rabbitmq:
    listener:
      simple:
        prefetch: 25 # Sweet spot cho hầu hết use case

// ❌ SAI — Tạo connection mỗi lần publish
public void badPublish(String message) {
    ConnectionFactory factory = new ConnectionFactory();
    Connection connection = factory.newConnection();  // Tốn kém!
    Channel channel = connection.createChannel();     // Mỗi lần!
    channel.basicPublish(...);
    channel.close();
    connection.close();
}
 
// ✅ ĐÚNG — Spring Boot quản lý connection pool
// RabbitTemplate tự động reuse connection và channel
@Autowired
private RabbitTemplate rabbitTemplate;
 
public void goodPublish(OrderEvent event) {
    rabbitTemplate.convertAndSend("exchange", "routing.key", event);
    // Connection và channel được pool và reuse
}

// Claim Check Pattern
public class ClaimCheckProducer {
 
    private final RabbitTemplate rabbitTemplate;
    private final S3Client s3Client;
 
    public void publishLargePayload(byte[] largePayload, String orderId) {
        // 1. Upload to S3
        String s3Key = "messages/" + orderId + "/" + UUID.randomUUID();
        s3Client.putObject(PutObjectRequest.builder()
                .bucket("message-payloads")
                .key(s3Key)
                .build(), RequestBody.fromBytes(largePayload));
 
        // 2. Publish reference only
        Map<String, String> claimCheck = Map.of(
                "orderId", orderId,
                "payloadRef", s3Key,
                "payloadSize", String.valueOf(largePayload.length)
        );
        rabbitTemplate.convertAndSend("order.exchange", "order.created", claimCheck);
    }
}

// Batch publish cho high-throughput scenarios
public void batchPublish(List<OrderEvent> events) {
    rabbitTemplate.invoke(operations -> {
        for (OrderEvent event : events) {
            operations.convertAndSend(
                    RabbitMQConfig.ORDER_EXCHANGE,
                    "order." + event.getEventType(),
                    event
            );
        }
        // Tất cả message được gửi trong cùng một channel operation
        // Giảm overhead so với gửi từng message
        return null;
    });
}

# application.yml — Enable RabbitMQ metrics với Micrometer
management:
  endpoints:
    web:
      exposure:
        include: health,metrics,prometheus
  metrics:
    tags:
      application: order-service
 
# Metrics quan trọng cần monitor:
# - rabbitmq.consumed: Số message consumed
# - rabbitmq.published: Số message published
# - rabbitmq.acknowledged: Số message acked
# - rabbitmq.rejected: Số message rejected
# - rabbitmq.channels: Số channels active
# - spring.rabbitmq.listener: Listener container metrics

# application.yml
management:
  endpoint:
    health:
      show-details: always
  health:
    rabbit:
      enabled: true # Auto-configured khi spring-boot-starter-amqp có mặt
 
 
# Response:
# {
#   "status": "UP",
#   "components": {
#     "rabbit": {
#       "status": "UP",
#       "details": {
#         "version": "3.12.0"
#       }
#     }
#   }
# }