RabbitMQ Overview, Installation, and Delayed Message Implementation Guide

What is RabbitMQ?

RabbitMQ is an open‑source implementation of the AMQP (Advanced Message Queuing Protocol) written in Erlang. It is lightweight, easy to deploy, supports multiple messaging protocols, and can be run in distributed or federated configurations for high scalability and availability.

Reliability: persistence, delivery acknowledgments, and publisher confirms.

Flexible Routing: exchanges route messages before they reach queues; built‑in and custom exchanges are available.

Clustering: multiple servers form a logical broker.

Highly Available Queues: mirrored queues survive node failures.

Multi‑protocol: supports STOMP, MQTT, etc.

Many Clients: libraries for Java, .NET, Ruby, and more.

Management UI: web interface for monitoring and administration.

Tracing: helps diagnose problematic messages.

Plugin System: extensible via plugins or custom development.

RabbitMQ Message Model

(Images illustrating exchanges, queues, and bindings are omitted for brevity.)

Why Use RabbitMQ?

Using a flower‑shop analogy, RabbitMQ solves three common problems:

Asynchronous Processing

Customers place orders by phone; the shop records the request and processes it later, allowing the customer to continue without waiting.

Decoupling

Instead of notifying each employee individually when a new order arrives, the order is written to a notebook; employees pick up tasks when they are free.

Load Shedding (Peak‑Cutting)

During peak periods, orders are queued and processed sequentially, preventing staff overload and ensuring a smooth workflow.

How to Install RabbitMQ

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

Enable the management plugin: rabbitmq-plugins enable rabbitmq_management

Verify installation at http://localhost:15672/

CentOS

Install Erlang (source compilation) and then RabbitMQ:

# Install Erlang (example)
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 3.8.16
export PATH=$PATH:/usr/local/rabbitmq_server-3.8.16/sbin
cd /usr/local/rabbitmq_server-3.8.16/sbin
./rabbitmq-server start

Implementing Delayed Messages

RabbitMQ can delay messages either via dead‑letter queues or the rabbitmq_delayed_message_exchange plugin. The following example uses the plugin with a Spring Boot application.

Prerequisites

Ensure the plugin version matches the RabbitMQ server version (e.g., rabbitmq_delayed_message_exchange for RabbitMQ 3.10.0).

pom.xml Dependency

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

application.yml Configuration

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

Java Configuration (Custom Exchange, Queue, Binding)

/**
 * Message queue configuration
 */
@Configuration
public class RabbitMqConfig {
    /**
     * Custom exchange for delayed messages
     */
    @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);
    }

    /**
     * Queue for delayed messages
     */
    @Bean
    public Queue chatPluginQueue() {
        return new Queue(QueueEnum.QUEUE_RESET_MESSAGE_CANCEL.getName());
    }

    /**
     * Bind queue to exchange
     */
    @Bean
    public Binding chatPluginBinding(CustomExchange chatPluginDirect, Queue chatPluginQueue) {
        return BindingBuilder.bind(chatPluginQueue)
                .to(chatPluginDirect)
                .with(QueueEnum.QUEUE_RESET_MESSAGE_CANCEL.getRouteKey())
                .noargs();
    }
}

Message Sender (sets x-delay header)

@Component
@Slf4j
public class ChatQueueSender {
    @Autowired
    private AmqpTemplate amqpTemplate;

    public void sendMessageToChat(Long cmid, final 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;
            }
        );
    }
}

Message 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 ChatQueueSender.sendMessageToChat(cmid, delay) wherever delayed processing is required.