rabbitmq pattern

// Direct exchange — route by exact key
@Bean
public DirectExchange orderExchange() {
   return new DirectExchange("order.exchange");
}
 
@Bean
public Queue orderCreatedQueue() {
   return QueueBuilder.durable("order.created.queue")
       .withArgument("x-dead-letter-exchange", "dlx.exchange")
       .withArgument("x-dead-letter-routing-key", "order.created.dlq")
       .build();
}
 
@Bean
public Binding orderBinding() {
   return BindingBuilder.bind(orderCreatedQueue())
       .to(orderExchange())
       .with("order.created"); // routing key
}
 
// Topic exchange — wildcard routing
@Bean
public TopicExchange eventExchange() {
   return new TopicExchange("event.exchange");
}
 
// Binds to "order.*" → matches order.created, order.cancelled, etc.
@Bean
public Binding auditBinding() {
   return BindingBuilder.bind(auditQueue())
       .to(eventExchange())
       .with("order.*");
}
 
// Fanout — all consumers receive
@Bean
public FanoutExchange notificationExchange() {
   return new FanoutExchange("notification.exchange");
}

@Service
@RequiredArgsConstructor
public class OrderEventProducer {
   private final RabbitTemplate rabbitTemplate;
 
   public void publishOrderCreated(Order order) {
       OrderCreatedEvent event = OrderCreatedEvent.builder()
           .orderId(order.getId())
           .timestamp(Instant.now())
           .build();
 
       rabbitTemplate.convertAndSend(
           "order.exchange",      // exchange
           "order.created",       // routing key
           event                  // message (auto-serialized via MessageConverter)
       );
   }
}

public void publishWithHeaders(Object payload, String traceId) {
   rabbitTemplate.convertAndSend("exchange", "routing.key", payload, message -> {
       MessageProperties props = message.getMessageProperties();
       props.setCorrelationId(traceId);
       props.setHeader("X-Source-Service", "order-service");
       props.setContentType("application/json");
       props.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
       return message;
   });
}

// Kích hoạt trong application.yml
// spring.rabbitmq.publisher-confirm-type: correlated
// spring.rabbitmq.publisher-returns: true
 
@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory factory) {
   RabbitTemplate template = new RabbitTemplate(factory);
   // Bắt buộc khi tự tạo RabbitTemplate: thiếu mandatory thì broker lặng lẽ
   // bỏ message không route được, ReturnsCallback không bao giờ được gọi
   template.setMandatory(true);
 
   // Xác nhận từ Broker
   template.setConfirmCallback((correlationData, ack, cause) -> {
       if (!ack) {
           log.error("Cảnh báo: Message chưa được confirm bởi Broker. Lý do: {}", cause);
           // Thực thi logic Retry hoặc lưu vào Database (Outbox/Failed Events table)
       }
   });
 
   // Bắt lỗi Unroutable messages
   template.setReturnsCallback(returned -> {
       log.error("Message bị trả về: routingKey={}, replyText={}",
           returned.getRoutingKey(), returned.getReplyText());
   });
   return template;
}

@Component
@Slf4j
public class OrderConsumer {
 
   // Simple consumer
   @RabbitListener(queues = "order.created.queue")
   public void handleOrderCreated(OrderCreatedEvent event) {
       log.info("Processing order: {}", event.getOrderId());
       processOrder(event);
   }
 
   // With message metadata
   @RabbitListener(queues = "order.created.queue")
   public void handleWithMetadata(OrderCreatedEvent event,
                                  @Header(AmqpHeaders.DELIVERY_TAG) long tag,
                                  @Header(AmqpHeaders.CORRELATION_ID) String correlationId,
                                  Channel channel) {
       log.info("[traceId={}] Processing order: {}", correlationId, event.getOrderId());
       processOrder(event);
   }
 
   // Batch consumer
   @RabbitListener(queues = "batch.queue", containerFactory = "batchListenerFactory")
   public void handleBatch(List<OrderCreatedEvent> events) {
       log.info("Processing batch of {} events", events.size());
       events.forEach(this::processOrder);
   }
}

@Bean
public SimpleRabbitListenerContainerFactory manualAckFactory(ConnectionFactory factory) {
   SimpleRabbitListenerContainerFactory f = new SimpleRabbitListenerContainerFactory();
   f.setConnectionFactory(factory);
   f.setAcknowledgeMode(AcknowledgeMode.MANUAL); // Kiểm soát hoàn toàn vòng đời message
   f.setPrefetchCount(10); // Không lấy quá 10 messages cùng lúc để tránh ngộp memory
   return f;
}
 
@RabbitListener(queues = "critical.queue", containerFactory = "manualAckFactory", concurrency = "5-20")
public void handleCriticalTask(Message message, Channel channel) throws IOException {
   long deliveryTag = message.getMessageProperties().getDeliveryTag();
   try {
       // Thực thi Business Logic
       processData(message);
 
       // Xác nhận thành công
       channel.basicAck(deliveryTag, false);
   } catch (RetryableBusinessException e) {
       // Lỗi có thể thử lại -> Requeue
       channel.basicNack(deliveryTag, false, true);
   } catch (Exception e) {
       // Lỗi nghiêm trọng (NPE, Invalid Data) -> Reject, đẩy thẳng xuống DLQ
       channel.basicNack(deliveryTag, false, false);
   }
}

// Multiple threads consuming from same queue
@RabbitListener(queues = "high-volume.queue", concurrency = "5-20")
public void handleConcurrent(Event event) {
   // 5 consumers min, scale up to 20 under load
}

@Bean
public Queue mainQueue() {
   return QueueBuilder.durable("order.queue")
       .withArgument("x-dead-letter-exchange", "dlx.exchange")
       .withArgument("x-dead-letter-routing-key", "order.dlq")
       .withArgument("x-message-ttl", 60000) // optional: message TTL
       .build();
}
 
@Bean
public DirectExchange dlxExchange() {
   return new DirectExchange("dlx.exchange");
}
 
@Bean
public Queue dlq() {
   return QueueBuilder.durable("order.dlq").build();
}
 
@Bean
public Binding dlqBinding() {
   return BindingBuilder.bind(dlq()).to(dlxExchange()).with("order.dlq");
}

@Bean
public SimpleRabbitListenerContainerFactory retryFactory(ConnectionFactory factory) {
   SimpleRabbitListenerContainerFactory f = new SimpleRabbitListenerContainerFactory();
   f.setConnectionFactory(factory);
   f.setAdviceChain(RetryInterceptorBuilder.stateless()
       .maxAttempts(3) // Tổng cộng 3 lần chạy: lần đầu + 2 lần retry
       .backOffOptions(1000, 2.0, 10000) // Chờ 1s rồi 2s giữa các lần (tối đa 10s)
       .recoverer(new RejectAndDontRequeueRecoverer()) // Hết 3 lần -> Bắn xuống DLQ
       .build());
   return f;
}

@Scheduled(fixedDelay = 300000) // Chạy mỗi 5 phút
public void reprocessDlq() {
   Message msg;
   int count = 0;
   // Lấy tối đa 100 messages mỗi lần chạy để tránh thắt nút cổ chai
   // Kiểm tra count TRƯỚC khi receive: receive() lấy message ra khỏi queue luôn (auto-ack),
   // đảo thứ tự điều kiện là message thứ 101 bị lấy ra rồi bỏ mất
   while (count < 100 && (msg = rabbitTemplate.receive("order.dlq")) != null) {
       try {
           rabbitTemplate.send("order.exchange", "order.created", msg);
           count++;
       } catch (Exception e) {
           log.error("Reprocess thất bại", e);
           // Message đã rời DLQ, gửi trả lại để không mất
           rabbitTemplate.send("dlx.exchange", "order.dlq", msg);
           break;
       }
   }
   if (count > 0) log.info("Đã khôi phục thành công {} messages từ DLQ", count);
}

@Component
@Slf4j
@RequiredArgsConstructor
public class IdempotentConsumer {
   private final StringRedisTemplate redis;
   private static final Duration DEDUP_TTL = Duration.ofHours(24);
 
   @RabbitListener(queues = "payment.queue")
   public void handle(PaymentEvent event) {
       String dedupKey = "processed:payment:" + event.getEventId();
 
       // Atomic operation: Chỉ set thành công nếu key chưa tồn tại
       Boolean isNew = redis.opsForValue().setIfAbsent(dedupKey, "1", DEDUP_TTL);
       if (Boolean.FALSE.equals(isNew)) {
           log.warn("Bỏ qua duplicate message: {}", event.getEventId());
           return;
       }
 
       try {
           processPayment(event);
       } catch (Exception e) {
           redis.delete(dedupKey); // Xóa key để có thể retry nếu thực sự xảy ra lỗi hệ thống
           throw e;
       }
   }
}

@Transactional
public void handleIdempotent(PaymentEvent event) {
   if (processedEventRepo.existsByEventId(event.getEventId())) {
       return; // already processed
   }
 
   processPayment(event);
 
   processedEventRepo.save(new ProcessedEvent(event.getEventId(), Instant.now()));
}

// Consistent hash — same orderId always goes to same queue/consumer
@Bean
public CustomExchange consistentHashExchange() {
   return new CustomExchange("order.hash.exchange", "x-consistent-hash", true, false);
}
 
// Binding with weight
@Bean
public Binding hashBinding1() {
   return BindingBuilder.bind(queue1())
       .to(consistentHashExchange())
       .with("1") // weight
       .noargs();
}

// 1. Core Logic (Write Side)
@Transactional
public Order handleCreateOrderCommand(CreateOrderCommand cmd) throws JsonProcessingException {
   // Lưu Aggregate Root
   Order order = orderRepo.save(new Order(cmd));
 
   // Lưu Event vào Outbox table trong CÙNG MỘT transaction
   outboxRepo.save(OutboxEvent.builder()
       .aggregateId(order.getId().toString())
       .eventType("ORDER_CREATED")
       .payload(objectMapper.writeValueAsString(order)) // JsonProcessingException là checked
       .status(OutboxStatus.PENDING)
       .build());
 
   return order;
}
 
// 2. Message Relay Worker
@Scheduled(fixedDelay = 1000)
@Transactional
public void publishPendingEvents() {
   List<OutboxEvent> events = outboxRepo.findTop100ByStatus(OutboxStatus.PENDING);
   for (OutboxEvent event : events) {
       try {
           // Payload đã là JSON: gửi thẳng bytes, đừng để Jackson converter serialize lần nữa
           // (sẽ thành một chuỗi JSON bị bọc trong dấu nháy)
           Message message = MessageBuilder
               .withBody(event.getPayload().getBytes(StandardCharsets.UTF_8))
               .setContentType(MessageProperties.CONTENT_TYPE_JSON)
               .build();
           rabbitTemplate.send("domain.events.exchange", event.getEventType(), message);
           event.setStatus(OutboxStatus.PUBLISHED);
       } catch (Exception e) {
           log.error("Không thể publish event: {}", event.getId());
       }
   }
   outboxRepo.saveAll(events);
}

// Service A: Thanh toán thành công -> Phát tín hiệu xác nhận đơn
@RabbitListener(queues = "payment.confirmed.queue")
public void onPaymentConfirmed(PaymentConfirmedEvent event) {
   orderService.confirmOrder(event.getOrderId());
   rabbitTemplate.convertAndSend("saga.exchange", "order.confirmed",
       new OrderConfirmedEvent(event.getOrderId()));
}
 
// Service A: Thanh toán thất bại -> Kích hoạt Compensating Transaction (Hoàn tác)
@RabbitListener(queues = "payment.failed.queue")
public void onPaymentFailed(PaymentFailedEvent event) {
   orderService.cancelOrder(event.getOrderId());
   rabbitTemplate.convertAndSend("saga.exchange", "order.cancelled",
       new OrderCancelledEvent(event.getOrderId()));
}

@Bean
public Jackson2JsonMessageConverter messageConverter() {
   ObjectMapper mapper = new ObjectMapper();
   mapper.registerModule(new JavaTimeModule());
   mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
   return new Jackson2JsonMessageConverter(mapper);
}
 
@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory factory, MessageConverter converter) {
   RabbitTemplate template = new RabbitTemplate(factory);
   template.setMessageConverter(converter);
   return template;
}

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtual-host: /
    publisher-confirm-type: correlated
    publisher-returns: true
    listener:
      simple:
        acknowledge-mode: auto
        prefetch: 10
        retry:
          enabled: true
          max-attempts: 3
          initial-interval: 1000ms
          multiplier: 2.0
          max-interval: 10000ms