RabbitMQ căn bản

RabbitMQ Broker
├── vhost: /production
│   ├── exchange: order.exchange
│   ├── queue: order.processing
│   └── queue: order.notification
├── vhost: /staging
│   ├── exchange: order.exchange
│   └── queue: order.processing
└── vhost: /development
    └── ...

// Các properties quan trọng thường dùng:
MessageProperties properties = MessagePropertiesBuilder.newInstance()
    .setContentType("application/json")       // Định dạng dữ liệu của body
    .setContentEncoding("UTF-8")              // Bảng mã hóa ký tự
    .setDeliveryMode(MessageDeliveryMode.PERSISTENT) // 1=tạm thời, 2=lưu đĩa (survive restart)
    .setPriority(5)                           // Độ ưu tiên, chỉ có tác dụng khi queue khai báo x-max-priority
    .setCorrelationId("abc-123")              // Dùng để định danh request-response (RPC)
    .setReplyTo("response.queue")             // Queue nhận phản hồi
    .setExpiration("60000")                   // TTL thời gian hết hạn (milliseconds)
    .setMessageId(UUID.randomUUID().toString()) // ID duy nhất chống lặp tin
    .setTimestamp(new Date())                 // Thời gian tạo tin nhắn
    .setType("OrderCreatedEvent")             // Tên loại sự kiện
    .setAppId("order-service")                // ID ứng dụng gửi tin
    .build();

// Ba loại xác nhận cơ bản:
channel.basicAck(deliveryTag, false);   // Xử lý thành công, cho phép xóa tin nhắn khỏi queue
channel.basicNack(deliveryTag, false, true);  // Thất bại, yêu cầu đẩy lại vào queue (requeue) để thử lại
channel.basicReject(deliveryTag, false);      // Thất bại, loại bỏ tin nhắn (hoặc chuyển sang DLX) không requeue

// 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");
 
// Gửi tin nhắn
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ị bỏ qua
 
// Publish — routing key đặt 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());

// Liên kết queue kèm điều kiện match header
Map<String, Object> headers = new HashMap<>();
headers.put("x-match", "all");  // "all" = tất cả điều kiện phải đúng (AND), "any" = chỉ cần 1 điều kiện (OR)
headers.put("format", "pdf");
headers.put("type", "report");
channel.queueBind("pdf-reports", "docs.headers", "", headers);
 
// Publish tin nhắn kèm header tương ứng
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: /
    # Timeout khi mở TCP connection tới broker
    connection-timeout: 5000
    # Listener configuration
    listener:
      simple:
        acknowledge-mode: manual # QUAN TRỌNG: Manual ack cho production
        prefetch: 10 # Số message tối đa chưa ack mà broker được đẩy tới mỗi consumer
        concurrency: 3 # Min consumer threads
        max-concurrency: 10 # Max consumer threads
        # Retry của Spring chỉ chạy khi listener ném exception ra ngoài. OrderConsumer bên dưới tự bắt
        # mọi exception rồi tự ack/nack, nên với code đó khối retry này không có tác dụng: chọn một trong hai cách.
        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 lombok.extern.slf4j.Slf4j;
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.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
@Slf4j
@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 ====================
 
    // Tự khai báo bean tên "rabbitListenerContainerFactory" thì Spring Boot sẽ KHÔNG tạo factory mặc định nữa.
    // Dùng configurer để vẫn áp các giá trị spring.rabbitmq.listener.simple.* trong application.yml
    // (acknowledge-mode, prefetch, concurrency, retry...), nếu không chúng bị bỏ qua.
    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
            SimpleRabbitListenerContainerFactoryConfigurer configurer,
            ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        configurer.configure(factory, connectionFactory);
        factory.setMessageConverter(jsonMessageConverter());
        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: mọi nhánh (kể cả nhánh lỗi) đều phải ack/nack/reject, nếu không message bị stuck ở trạng thái unacked
     */
    @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:4-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

+----------+  publish -->  +----------+
| Producer |               | Exchange |
+----------+  <-- confirm  +----------+

@Slf4j
@Service
@RequiredArgsConstructor
public class ReliableProducer {
 
    private final RabbitTemplate rabbitTemplate;
    private final OrderOutboxRepository outboxRepository; // Outbox pattern
    private final ObjectMapper objectMapper;
 
    /**
     * Transactional Outbox Pattern:
     * 1. Save business data + bản ghi outbox trong CÙNG một DB transaction
     * 2. KHÔNG publish ngay trong transaction: nếu publish xong mà transaction rollback,
     *    consumer sẽ nhận event của một đơn hàng không hề tồn tại
     * 3. Một relay (scheduler bên dưới, hoặc CDC như Debezium) đọc outbox rồi publish
     * 4. Chỉ mark SENT khi broker đã confirm
     */
    @Transactional
    public void saveWithOutbox(OrderEvent event) throws JsonProcessingException {
        // orderRepository.save(...);  // business data, cùng transaction
 
        OrderOutbox outbox = OrderOutbox.builder()
                .messageId(UUID.randomUUID().toString())
                .exchange(RabbitMQConfig.ORDER_EXCHANGE)
                .routingKey("order." + event.getEventType())
                .payload(objectMapper.writeValueAsString(event))
                .status(OutboxStatus.PENDING)
                .retryCount(0)
                .createdAt(LocalDateTime.now())
                .build();
        outboxRepository.save(outbox);
    }
 
    /**
     * Relay: publish các bản ghi PENDING, chạy mỗi 5 giây.
     * Chạy nhiều instance thì phải khoá bản ghi (SELECT ... FOR UPDATE SKIP LOCKED, ShedLock...)
     * để hai instance không cùng gửi một message.
     */
    @Scheduled(fixedDelay = 5000)
    public void relayPendingMessages() {
        List<OrderOutbox> pendingMessages = outboxRepository
                .findTop100ByStatusOrderByCreatedAtAsc(OutboxStatus.PENDING);
 
        for (OrderOutbox outbox : pendingMessages) {
            CorrelationData correlationData = new CorrelationData(outbox.getMessageId());
            try {
                rabbitTemplate.convertAndSend(
                        outbox.getExchange(),
                        outbox.getRoutingKey(),
                        objectMapper.readValue(outbox.getPayload(), OrderEvent.class),
                        message -> {
                            // Giữ nguyên messageId qua mọi lần gửi lại để consumer khử trùng được
                            message.getMessageProperties().setMessageId(outbox.getMessageId());
                            return message;
                        },
                        correlationData
                );
 
                // Chờ publisher confirm (cần publisher-confirm-type: correlated)
                CorrelationData.Confirm confirm = correlationData.getFuture().get(10, TimeUnit.SECONDS);
                if (confirm.isAck() && correlationData.getReturned() == null) {
                    outbox.setStatus(OutboxStatus.SENT);
                } else {
                    // nack từ broker, hoặc message không route được tới queue nào
                    outbox.setRetryCount(outbox.getRetryCount() + 1);
                }
            } catch (Exception e) {
                log.warn("Publish failed: {}", outbox.getMessageId(), e);
                outbox.setRetryCount(outbox.getRetryCount() + 1);
            }
 
            if (outbox.getStatus() != OutboxStatus.SENT && 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);
        }
    }
}

              +-------------+
              | Main Queue  |
              |  (order.q)  |
              +------+------+
                     |
              Consumer reject
              hoac TTL expired
                     |
                     v
              +-------------+
              | DLX Exchange|
              +------+------+
                     |
                     v
              +-------------+
              |  DLX Queue  |--> DLX Consumer
              |  (parking)  |    (analyze, retry, alert)
              +-------------+

@Slf4j
@Component
@RequiredArgsConstructor
public class DeadLetterConsumer {
 
    // Header tự quản lý để đếm số lần retry (xem giải thích bên dưới)
    private static final String RETRY_HEADER = "x-retry-count";
 
    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);              // Lần dead letter gần nhất
            String reason = (String) deathInfo.get("reason");       // "rejected", "expired", "maxlen", "delivery_limit"
            String queue = (String) deathInfo.get("queue");          // Queue gốc
            String exchange = (String) deathInfo.get("exchange");    // Exchange gốc
            // Routing key gốc. Không dùng receivedRoutingKey: nó đã bị thay bằng x-dead-letter-routing-key ("dlx.routing")
            List<String> routingKeys = (List<String>) deathInfo.get("routing-keys");
            String originalRoutingKey = routingKeys.get(0);
 
            // Từ RabbitMQ 4.0, broker không đọc x-death do client gửi kèm khi publish lại,
            // nên "count" trong x-death không tăng qua các vòng retry kiểu này. Tự đếm bằng header riêng.
            int retryCount = ((Number) headers.getOrDefault(RETRY_HEADER, 0)).intValue();
 
            log.error("Dead letter received: reason={}, queue={}, retryCount={}, exchange={}",
                    reason, queue, retryCount, exchange);
 
            // Retry logic dựa trên số lần fail
            if (retryCount < 3) {
                // Retry — gửi lại queue gốc sau delay
                log.info("Retrying message (attempt {}): queue={}", retryCount + 1, queue);
 
                // Delay trước khi retry (exponential backoff).
                // Lưu ý: sleep chặn luôn thread của listener, xem cách tốt hơn ở mục delay queue bên dưới.
                try {
                    Thread.sleep(Math.min(1000 * (long) Math.pow(2, retryCount), 30000));
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
 
                message.getMessageProperties().setHeader(RETRY_HEADER, retryCount + 1);
                rabbitTemplate.send(exchange, originalRoutingKey, message);
            } else {
                // Đã retry quá nhiều — alert và archive
                log.error("Message permanently failed after {} retries. Alerting team.", retryCount);
                alertService.sendAlert(
                        "Dead Letter Alert",
                        String.format("Message failed %d times in queue %s. Manual intervention needed.",
                                retryCount, 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.
    // Queue chính phải dead-letter với đúng routing key này (x-dead-letter-routing-key = "retry.order"),
    // nếu vẫn để "dlx.routing" như RabbitMQConfig ở trên thì message rơi vào dlx.queue, không bao giờ tới retry queue.
    @Bean
    public Binding retryBinding() {
        return BindingBuilder
                .bind(retryQueue())
                .to(new DirectExchange(RabbitMQConfig.DLX_EXCHANGE))
                .with("retry.order");
    }
}

Flow: Main Queue -> reject -> DLX -> Retry Queue (wait 5s) -> TTL expire -> Main Queue
                                                                            |
                                                                            v
                                                                      Consumer retry

# 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 dùng CachingConnectionFactory
// Mặc định: một connection dùng chung + cache channel, RabbitTemplate tự reuse
@Autowired
private RabbitTemplate rabbitTemplate;
 
public void goodPublish(OrderEvent event) {
    rabbitTemplate.convertAndSend("exchange", "routing.key", event);
    // Connection được dùng chung, channel lấy từ cache rồi trả lại
}

// 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 # tiện khi dev; production nên dùng when-authorized
  health:
    rabbit:
      enabled: true # Auto-configured khi spring-boot-starter-amqp có mặt
 
 
# Response:
# {
#   "status": "UP",
#   "components": {
#     "rabbit": {
#       "status": "UP",
#       "details": {
#         "version": "4.1.0"
#       }
#     }
#   }
# }