Chapter 3: Asynchronous Messaging with RabbitMQ

Chapter Objectives

• Understand asynchronous messaging concepts
• Configure RabbitMQ with Spring Boot
• Implement producers and consumers
• Manage exchanges, queues, and bindings
• Implement messaging patterns

---

1. Introduction to Messaging

1.1 Why Asynchronous Messaging

Synchronous (REST):
Service A --HTTP--> Service B --HTTP--> Service C
    └── Wait for response ──┘     └── Wait for response ──┘
    
Problems:
- Temporal coupling
- Service B down = Service A fails
- Cumulative latency

Asynchronous (Messaging):
Service A --publish--> [Queue] --consume--> Service B
    └── Continue immediately

Advantages:
- Decoupling
- Resilience
- Scalability
- Peak handling

1.2 RabbitMQ Concepts

Concept	Description
Producer	Sends messages
Consumer	Receives messages
Queue	Message waiting line
Exchange	Routes messages to queues
Binding	Routing rule exchange -> queue
Routing Key	Key for routing

1.3 Types of Exchange

Direct Exchange:
Producer --> [Exchange] --> Queue (routing_key exact match)

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

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

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

---

2. Spring AMQP Configuration

2.1 Maven Dependencies

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

2.2 application.yml Configuration

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 Java Configuration

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 {
    
    // Queue and exchange names
    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 for orders
    @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();
    }
    
    // Direct Exchange for orders
    @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 for notifications
    @Bean
    public Queue notificationQueue() {
        return QueueBuilder.durable(NOTIFICATION_QUEUE).build();
    }
    
    // Fanout Exchange for notifications (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);
    }
    
    // JSON Converter
    @Bean
    public MessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }
    
    // Configured template
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        template.setMessageConverter(jsonMessageConverter());
        return template;
    }
}

---

3. Producer (Sending Messages)

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 and 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 Producer Service

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("Sending order created event: {}", event);
        
        rabbitTemplate.convertAndSend(
            RabbitMQConfig.ORDER_EXCHANGE,
            RabbitMQConfig.ORDER_ROUTING_KEY,
            event
        );
        
        log.info("Order event sent successfully");
    }
    
    public void sendNotification(String message) {
        log.info("Broadcasting notification: {}", message);
        
        // Fanout - no routing key needed
        rabbitTemplate.convertAndSend(
            RabbitMQConfig.NOTIFICATION_EXCHANGE,
            "",
            message
        );
    }
    
    // Send with confirmation
    public void sendWithConfirmation(OrderEvent event) {
        rabbitTemplate.invoke(operations -> {
            operations.convertAndSend(
                RabbitMQConfig.ORDER_EXCHANGE,
                RabbitMQConfig.ORDER_ROUTING_KEY,
                event
            );
            
            // Wait for confirmation
            boolean confirmed = operations.waitForConfirms(5000);
            if (!confirmed) {
                throw new RuntimeException("Message not confirmed by broker");
            }
            
            return null;
        });
    }
}

3.3 Usage in a 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. Create the order
        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. Publish the event
        OrderEvent event = new OrderEvent(
            saved.getId(),
            saved.getUserId(),
            saved.getProductName(),
            saved.getTotalAmount(),
            saved.getStatus()
        );
        
        eventProducer.sendOrderCreatedEvent(event);
        
        return saved;
    }
}

---

4. Consumer (Receiving Messages)

4.1 Simple Listener

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("Received order event: {}", event);
        
        try {
            // Message processing
            processOrder(event);
            log.info("Order event processed successfully");
        } catch (Exception e) {
            log.error("Error processing order event: {}", e.getMessage());
            throw e; // Re-throw for retry/DLQ
        }
    }
    
    private void processOrder(OrderEvent event) {
        // Business logic
        // - Send confirmation email
        // - Update stock
        // - Notify delivery system
        log.info("Processing order {} for user {}", 
            event.getOrderId(), event.getUserId());
    }
}

4.2 Listener with Manual Acknowledgment

@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("Received order event: {}", event);
        
        try {
            processOrder(event);
            
            // Acknowledge success
            channel.basicAck(deliveryTag, false);
            log.info("Message acknowledged");
            
        } catch (Exception e) {
            log.error("Error processing: {}", e.getMessage());
            
            try {
                // Reject and requeue (or send to DLQ)
                channel.basicNack(deliveryTag, false, false);
            } catch (IOException ioEx) {
                log.error("Error rejecting message", ioEx);
            }
        }
    }
}

4.3 Listener for Notifications (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("Received notification: {}", message);
        // Process the notification
    }
}

4.4 Multiple Listeners (Scaling)

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

---

5. Dead Letter Queue (DLQ)

5.1 DLQ Configuration

@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 DLQ Listener

@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 received: {}", event);
        
        if (xDeath != null && !xDeath.isEmpty()) {
            Map<String, Object> death = xDeath.get(0);
            log.error("Death reason: {}, count: {}", 
                death.get("reason"), death.get("count"));
        }
        
        // Possible actions:
        // 1. Save for analysis
        // 2. Notify ops team
        // 3. Manual retry
        
        notificationService.alertOpsTeam(
            "Order processing failed: " + event.getOrderId()
        );
    }
}

---

6. Topic Exchange (Advanced Routing)

6.1 Configuration

@Configuration
public class TopicExchangeConfig {
    
    public static final String TOPIC_EXCHANGE = "events.topic";
    
    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange(TOPIC_EXCHANGE);
    }
    
    // Queue for all order events
    @Bean
    public Queue allOrdersQueue() {
        return new Queue("queue.orders.all");
    }
    
    // Queue for created orders only
    @Bean
    public Queue createdOrdersQueue() {
        return new Queue("queue.orders.created");
    }
    
    // Queue for all events
    @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("#");  // All messages
    }
}

6.2 Usage

@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 Configuration

@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 RPC Client

@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 RPC Server

@Component
public class RpcServer {
    
    @RabbitListener(queues = RpcConfig.RPC_QUEUE)
    public String handleRequest(String request) {
        log.info("Received RPC request: {}", request);
        
        // Processing
        String result = processRequest(request);
        
        log.info("Sending RPC response: {}", result);
        return result;
    }
    
    private String processRequest(String request) {
        // Business logic
        return "Processed: " + request;
    }
}

---

8. Summary

Messaging Patterns

Pattern	Exchange	Usage
Point-to-point	Direct	One consumer
Publish/Subscribe	Fanout	Broadcast
Routing	Topic	Pattern-based routing
Request/Reply	Direct	RPC

Best Practices

1. Use idempotent messages
2. Implement a Dead Letter Queue
3. Configure appropriate retries
4. Monitor queues (RabbitMQ Management)
5. Use JSON serialization
6. Document message formats
7. Test with invalid messages

---

Next Chapter: Scheduled Tasks

Back to Module 14