Master Spring Boot & RabbitMQ: 4 Core Messaging Patterns & Production Optimizations

Overview

In modern distributed systems, a message queue (MQ) is a key component for decoupling services, smoothing traffic spikes, and handling asynchronous processing. This article demonstrates how to use Spring Boot + RabbitMQ to implement four core messaging scenarios and applies production‑grade optimizations.

Four Core Scenarios

Simple Queue (point‑to‑point)

Work Queue (competing consumers)

Publish/Subscribe (fanout broadcast)

Routing (direct exchange)

Each scenario is enhanced with the following production considerations:

Message persistence

Manual ACK

Prefetch limit

Dead‑letter exchange (DLX)

Publisher confirms and returns

Idempotency and retry mechanisms

Project Configuration

1. Maven Dependencies

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>com.fasterxml.jackson.core</groupId>
        <artifactId>jackson-databind</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.retry</groupId>
        <artifactId>spring-retry</artifactId>
    </dependency>
</dependencies>

2. application.yml

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtual-host: /
    publisher-confirm-type: correlated   # enable publisher confirms
    publisher-returns: true              # enable returns
    listener:
      simple:
        acknowledge-mode: manual      # manual ACK
        prefetch: 1                   # fetch one message at a time for fair dispatch

3. Common Configuration Class

@Configuration
public class RabbitConfig {
    // JSON message converter
    @Bean
    public MessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory, MessageConverter converter) {
        RabbitTemplate rt = new RabbitTemplate(connectionFactory);
        rt.setMessageConverter(converter);
        rt.setMandatory(true);
        return rt;
    }

    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory, MessageConverter converter) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(converter);
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        factory.setPrefetchCount(1);
        return factory;
    }
}

Implementation of Core Scenarios

1. Simple Queue (point‑to‑point)

Queue Definition

@Bean
public Queue simpleQueue() {
    return new Queue("simple.queue", true);
}

Producer

@Service
public class SimpleQueueProducer {
    private final RabbitTemplate rabbitTemplate;
    public SimpleQueueProducer(RabbitTemplate rabbitTemplate) { this.rabbitTemplate = rabbitTemplate; }
    public void sendMessage(String msg) {
        rabbitTemplate.convertAndSend("simple.queue", msg);
        System.out.println("Sent message: " + msg);
    }
}

Consumer

@Component
@RabbitListener(queues = "simple.queue")
public class SimpleQueueConsumer {
    @RabbitHandler
    public void receiveMessage(Message message, Channel channel) throws IOException {
        try {
            String body = new String(message.getBody(), StandardCharsets.UTF_8);
            System.out.println("Received message: " + body);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
        }
    }
}

2. Work Queue (competing consumers)

Queue Definition

@Bean
public Queue workQueue() {
    return new Queue("work.queue", true);
}

Producer

@Service
public class WorkQueueProducer {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    public void sendMessage(String msg) {
        rabbitTemplate.convertAndSend("work.queue", msg);
    }
}

Consumer (multiple instances)

@Component
@RabbitListener(queues = "work.queue", concurrency = "2-5")
public class WorkQueueConsumer {
    @RabbitHandler
    public void process(String msg) throws InterruptedException {
        System.out.println("Processing task: " + msg);
        Thread.sleep(200);
    }
}

3. Publish/Subscribe (Fanout Exchange)

Exchange, Queues and Bindings

@Bean
public FanoutExchange fanoutExchange() {
    return new FanoutExchange("fanout.exchange");
}
@Bean
public Queue fanoutQueue1() { return new Queue("fanout.queue1", true); }
@Bean
public Queue fanoutQueue2() { return new Queue("fanout.queue2", true); }
@Bean
public Binding binding1(FanoutExchange e, Queue q1) { return BindingBuilder.bind(q1).to(e); }
@Bean
public Binding binding2(FanoutExchange e, Queue q2) { return BindingBuilder.bind(q2).to(e); }

Producer

@Service
public class FanoutProducer {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    public void sendMessage(String msg) {
        rabbitTemplate.convertAndSend("fanout.exchange", "", msg);
    }
}

4. Routing (Direct Exchange)

Exchange, Queues and Bindings

@Bean
public DirectExchange directExchange() { return new DirectExchange("direct.exchange"); }
@Bean
public Queue qError() { return new Queue("direct.queue1", true); }
@Bean
public Queue qInfoWarn() { return new Queue("direct.queue2", true); }
@Bean
public Binding bError(DirectExchange e, Queue qError) { return BindingBuilder.bind(qError).to(e).with("error"); }
@Bean
public Binding bInfo(DirectExchange e, Queue qInfoWarn) { return BindingBuilder.bind(qInfoWarn).to(e).with("info"); }
@Bean
public Binding bWarn(DirectExchange e, Queue qInfoWarn) { return BindingBuilder.bind(qInfoWarn).to(e).with("warning"); }

Producer

@Service
public class DirectProducer {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    public void send(String type, String msg) {
        rabbitTemplate.convertAndSend("direct.exchange", type, msg);
    }
}

Controller and Testing

@RestController
@RequestMapping("/mq")
public class MQController {
    @Autowired private SimpleQueueProducer simpleQueueProducer;
    @Autowired private WorkQueueProducer workQueueProducer;
    @Autowired private FanoutProducer fanoutProducer;
    @Autowired private DirectProducer directProducer;

    @PostMapping("/simple")
    public String simple(@RequestParam String message) {
        simpleQueueProducer.sendMessage(message);
        return "Simple queue message sent";
    }

    @PostMapping("/work")
    public String work(@RequestParam String message, @RequestParam int count) {
        for (int i = 0; i < count; i++) {
            workQueueProducer.sendMessage(message + "-" + i);
        }
        return "Work queue messages sent";
    }

    @PostMapping("/fanout")
    public String fanout(@RequestParam String message) {
        fanoutProducer.sendMessage(message);
        return "Fanout broadcast sent";
    }

    @PostMapping("/direct")
    public String direct(@RequestParam String type, @RequestParam String message) {
        directProducer.send(type, message);
        return "Direct routing message sent";
    }
}

Test commands (using curl):

# Simple queue
curl -X POST "http://localhost:8080/mq/simple?message=HelloSimple"

# Work queue (10 messages)
curl -X POST "http://localhost:8080/mq/work?message=Task&count=10"

# Fanout broadcast
curl -X POST "http://localhost:8080/mq/fanout?message=Broadcast"

# Direct routing (error type)
curl -X POST "http://localhost:8080/mq/direct?type=error&message=ErrorOccurred"

Advanced Features and Optimizations

Message Persistence – queues, exchanges and messages are durable to survive broker restarts.

Manual ACK + Dead‑Letter Queue – on failure, basicNack routes the message to a DLX.

Prefetch Limit – ensures fair dispatch and prevents slow consumers from blocking the queue.

Publisher Confirms + Returns – verify that messages reach the broker/queue, avoiding loss.

Idempotency & Retry – design business logic to be idempotent to handle duplicate deliveries.

Monitoring & Alerting – enable RabbitMQ management plugin, monitor queue depth and consumer health, integrate with Prometheus/Grafana for alerts.

Architecture Diagram

Conclusion

The article fully implements the four core RabbitMQ scenarios—simple queue, work queue, publish/subscribe, and routing—within a Spring Boot application, and augments them with production‑grade features such as persistence, manual ACK, prefetch control, dead‑letter handling, idempotent processing, and monitoring. This architecture provides a scalable, highly reliable messaging backbone suitable for e‑commerce orders, log processing, notifications, and other typical business use cases.