RabbitMQ Overview, Installation Guide, and Delayed Message Implementation

Hello, I am the guide author.

What is RabbitMQ?

RabbitMQ is an open‑source implementation of the AMQP (Advanced Message Queuing Protocol) developed in Erlang. It is lightweight and easy to deploy, supporting multiple messaging protocols. It can be deployed in distributed and federated configurations to meet high‑scale, high‑availability requirements.

Key characteristics include:

Reliability : persistence, delivery acknowledgments, and publisher confirms.

Flexible Routing : exchanges route messages before they enter queues; built‑in exchanges handle typical routing, while custom exchanges can be created via plugins.

Clustering : multiple RabbitMQ servers form a logical broker.

Highly Available Queues : queues can be mirrored across cluster nodes.

Multi‑protocol Support : STOMP, MQTT, etc.

Many Language Clients : Java, .NET, Ruby, and more.

Management UI : a user‑friendly interface for monitoring and managing the broker.

Tracing : helps identify issues with messages.

Plugin System : extensible via plugins, including custom ones.

RabbitMQ Message Model

(Image illustrating the model)

Why Use RabbitMQ?

Illustrated with a flower‑shop scenario:

Asynchronous

Customers place orders and wait while staff prepares them; the customer can only leave after the order is ready.

Decoupling

When a new order arrives, the owner notifies each staff member individually; adding or removing staff only changes the notification list.

With RabbitMQ, the owner simply records the order in a notebook; staff pick up tasks when they are free.

Throttling (Peak‑shaving)

During peak periods (e.g., a holiday), many orders flood in and staff become overwhelmed.

By acknowledging orders and processing them in order, or encouraging early purchases, the load is smoothed.

How to Use RabbitMQ?

Basic Installation Methods

macOS

brew install rabbitmq

Windows

Install Erlang: http://erlang.org/download/otp_win64_21.3.exe

Install RabbitMQ: https://dl.bintray.com/rabbitmq/all/rabbitmq-server/3.7.14/rabbitmq-server-3.7.14.exe

After installation, enable the management plugin:

rabbitmq-plugins enable rabbitmq_management

Verify the installation by visiting http://localhost:15672/ .

CentOS

Install Erlang (source compilation) and then RabbitMQ:

# Install Erlang (example commands)
cd /path/to/erlang-source && ./configure --prefix=/usr/local/erlang
make && make install
export PATH=$PATH:/usr/local/erlang/bin
source /etc/profile
# Verify with 'erl'

# Install RabbitMQ (example commands)
# Download rabbitmq-server-3.8.16 and move to /usr/local/
export PATH=$PATH:/usr/local/rabbitmq_server-3.8.16/sbin
source /etc/profile
cd /usr/local/rabbitmq_server-3.8.16/sbin
./rabbitmq-server start

Feature Implementation – Delayed Messages

RabbitMQ supports delayed messages via dead‑letter queues or the rabbitmq_delayed_message_exchange plugin; this guide uses the plugin.

Alternative Installation (Recommended)

Download matching versions of Erlang and RabbitMQ (e.g., erl_25.0 and rabbitmq‑3.10.0) and the rabbitmq_delayed_message_exchange plugin.

Copy the plugin file into the plugins directory and enable it:

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

Implementing Delayed Messages

Example: if a customer service agent is online but does not reply within 3 minutes, a bot takes over the conversation.

Add AMQP dependency to pom.xml :

org.springframework.boot
spring-boot-starter-amqp

Configure RabbitMQ in application.yml :

spring:
  rabbitmq:
    host: localhost
    port: 5672
    virtual-host: /mall
    username: im
    password: xxxxxx
    publisher-confirms: true

Create a Java configuration class to define a custom delayed exchange, queue, and binding:

/** Message queue configuration */
@Configuration
public class RabbitMqConfig {
    @Bean
    CustomExchange chatPluginDirect() {
        Map
args = new HashMap<>();
        args.put("x-delayed-type", "direct");
        return new CustomExchange(QueueEnum.QUEUE_RESET_MESSAGE_CANCEL.getExchange(), "x-delayed-message", true, false, args);
    }
    @Bean
    public Queue chatPluginQueue() {
        return new Queue(QueueEnum.QUEUE_RESET_MESSAGE_CANCEL.getName());
    }
    @Bean
    public Binding chatPluginBinding(CustomExchange chatPluginDirect, Queue chatPluginQueue) {
        return BindingBuilder.bind(chatPluginQueue).to(chatPluginDirect).with(QueueEnum.QUEUE_RESET_MESSAGE_CANCEL.getRouteKey()).noargs();
    }
}

Sender that sets the x-delay header:

/** Robot restart queue sender */
@Component
@Slf4j
public class ChatQueueSender {
    @Autowired
    private AmqpTemplate amqpTemplate;
    public void sendMessageToChat(Long cmid, long delayTimes) {
        amqpTemplate.convertAndSend(QueueEnum.QUEUE_RESET_MESSAGE_CANCEL.getExchange(),
            QueueEnum.QUEUE_RESET_MESSAGE_CANCEL.getRouteKey(), cmid, message -> {
                message.getMessageProperties().setHeader("x-delay", delayTimes);
                return message;
            });
    }
}

Receiver that processes delayed messages:

/** Robot restart queue receiver */
@Component
@Slf4j
@RabbitListener(queues = "im.chat.cancel")
public class ChatQueueReceiver {
    @Autowired
    private ChatRestartRobotService chatRestartRobotService;
    @RabbitHandler
    public void handleOnChat(Long cmid) {
        chatRestartRobotService.restartRobot(cmid);
    }
}

Invoke the sender where needed to schedule a delayed task.

Done!