Master RabbitMQ: Core Concepts, Programming Models, and Spring Boot Integration

RabbitMQ Overview

RabbitMQ is an open‑source, AMQP‑based enterprise messaging system.

It runs on Linux, Windows, macOS and supports Java, Python, Ruby, .NET, PHP, C/C++, Node.js, etc.

AMQP (Advanced Message Queuing Protocol) is an open standard network protocol for message‑oriented middleware, released in 2006. RabbitMQ 1.0, built on Erlang, was released in 2007.

RabbitMQ Core Concepts

Broker

The RabbitMQ server acts as a message broker, receiving and routing messages.

Virtual Host

A virtual host (vhost) isolates resources for multi‑tenant security, providing a separate namespace for exchanges, queues, and bindings.

Connection

Clients establish a TCP connection to RabbitMQ; multiple channels can be multiplexed over a single connection for better performance.

Channel

Each channel is an AMQP session with a unique identifier, used for publishing and consuming.

Exchange

Messages are first sent to an exchange, which routes them to queues based on routing rules. Common exchange types are fanout, direct, topic, and headers.

Queue

Queues store messages in FIFO order until a consumer retrieves them.

Producer

A producer sends messages to an exchange (or directly to a queue via the default exchange).

Consumer

A consumer receives messages from a queue.

Message Flow Steps

Producer steps:

Create a producer and define a group name.

Specify the Nameserver address.

Start the producer.

Create a message object with topic, tag, and body.

Send the message.

Close the producer.

Consumer steps:

Create a consumer and define a group name.

Specify the Nameserver address.

Subscribe to a topic and tag.

Set a callback to process messages.

Start the consumer.

Programming Model (Java)

Dependency

<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.9.0</version>
</dependency>

Create Connection and Channel

ConnectionFactory factory = new ConnectionFactory();
factory.setHost(HOST_NAME);
factory.setPort(HOST_PORT);
factory.setUsername(USER_NAME);
factory.setPassword(PASSWORD);
factory.setVirtualHost(VIRTUAL_HOST);
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();

Declare Exchange (optional)

channel.exchangeDeclare(String exchange, String type, boolean durable, boolean autoDelete, Map<String, Object> arguments) throws IOException;

Exchange types: fanout, direct, topic, headers.

Declare Queue

channel.queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments);

durable

ensures messages survive broker restarts.

Bind Exchange and Queue (optional)

channel.queueBind(String queue, String exchange, String routingKey) throws IOException;

Publish Message

channel.basicPublish(String exchange, String routingKey, BasicProperties props, message.getBytes("UTF-8"));

If no exchange is needed, use an empty string for the exchange name.

Message Properties

AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
builder.deliveryMode(MessageProperties.PERSISTENT_TEXT_PLAIN.getDeliveryMode());
builder.priority(MessageProperties.PERSISTENT_TEXT_PLAIN.getPriority());
AMQP.BasicProperties prop = builder.build();

MessageProperties.PERSISTENT_TEXT_PLAIN

provides default persistent settings.

Consume Message

Passive Consumption

channel.basicConsume(String queue, boolean autoAck, Consumer callback);

Active Consumption

GetResponse response = channel.basicGet(QUEUE_NAME, boolean autoAck);

Acknowledgment

Auto‑ACK (autoAck=true): the broker marks the message as delivered immediately.

Manual ACK (autoAck=false): the consumer must call channel.basicAck to confirm processing, preventing loss on failure.

Resource Cleanup

channel.close();
connection.close();

Messaging Models

Simple Model

Producer sends directly to a queue; consumer reads from that queue. No exchange is defined.

Work Queue

Multiple consumers share a single queue; each message is delivered to only one consumer, improving throughput for heavy tasks.

Publish/Subscribe (Fanout)

Producer sends to a fanout exchange; the exchange broadcasts to all bound queues.

Routing (Direct)

Messages are routed to queues whose binding key exactly matches the routing key.

Topic (Wildcard)

Using a topic exchange, routing keys can contain * (single word) and # (zero or more words) wildcards.

Message Confirmation

Enable publisher confirms to avoid lost messages: channel.confirmSelect(); Register a confirm listener:

channel.addConfirmListener(ConfirmCallback ackCallback, ConfirmCallback nackCallback);

Retrieve the sequence number for each published message:

long seqNo = channel.getNextPublishSeqNo();

Spring Boot Integration

Add Dependency

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

Configuration (application.yml)

server:
  port: 8081
spring:
  application:
    name: test-rabbitmq-producer
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest
    virtualHost: /

RabbitMQ Config Class

package com.example.config;

import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitmqConfig {
    public static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
    public static final String QUEUE_INFORM_SMS = "queue_inform_sms";
    public static final String EXCHANGE_TOPICS_INFORM = "exchange_topics_inform";
    public static final String ROUTINGKEY_EMAIL = "inform.#.email.#";
    public static final String ROUTINGKEY_SMS = "inform.#.sms.#";

    @Bean(EXCHANGE_TOPICS_INFORM)
    public Exchange EXCHANGE_TOPICS_INFORM(){
        return ExchangeBuilder.topicExchange(EXCHANGE_TOPICS_INFORM).durable(true).build();
    }

    @Bean(QUEUE_INFORM_EMAIL)
    public Queue QUEUE_INFORM_EMAIL(){
        return new Queue(QUEUE_INFORM_EMAIL);
    }

    @Bean(QUEUE_INFORM_SMS)
    public Queue QUEUE_INFORM_SMS(){
        return new Queue(QUEUE_INFORM_SMS);
    }

    @Bean
    public Binding BINDING_QUEUE_INFORM_EMAIL(@Qualifier(QUEUE_INFORM_EMAIL) Queue queue,
                                              @Qualifier(EXCHANGE_TOPICS_INFORM) Exchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with(ROUTINGKEY_EMAIL).noargs();
    }

    @Bean
    public Binding BINDING_ROUTINGKEY_SMS(@Qualifier(QUEUE_INFORM_SMS) Queue queue,
                                          @Qualifier(EXCHANGE_TOPICS_INFORM) Exchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with(ROUTINGKEY_SMS).noargs();
    }
}

Send Message

String message = "hello world";
rabbitTemplate.convertAndSend(RabbitmqTopicConfig.EXCHANGE_TOPICS_INFORM, "inform.email", message);

Consume Message

@RabbitListener(queues = {RabbitmqTopicConfig.QUEUE_INFORM_EMAIL})
public void receive_email(Object msg, Message message, Channel channel){
    System.out.println("QUEUE_INFORM_EMAIL msg" + msg);
}

@RabbitListener(queues = {RabbitmqTopicConfig.QUEUE_INFORM_SMS})
public void receive_sms(Object msg, Message message, Channel channel){
    System.out.println("QUEUE_INFORM_SMS msg" + msg);
}