Implementing Delayed Messaging with RabbitMQ Plugin: Installation, Configuration, and Java Example

RabbitMQ is an open‑source AMQP implementation written in Erlang, known for its reliability, flexible routing, clustering, high availability, multi‑protocol support, and extensive plugin ecosystem.

Using RabbitMQ helps solve common problems such as asynchronous processing, decoupling services, and load‑leveling (peak shaving) by allowing producers to publish messages without waiting for consumers.

Installation

Mac :

brew install rabbitmq

Windows (install Erlang first, then RabbitMQ – see the original article for download links).

After installation, enable the management UI:

rabbitmq-plugins enable rabbitmq_management

To use delayed messages, enable the delayed‑message plugin:

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

CentOS installation example

# Install Erlang (example commands)
cd /path/to/erlang-source && ./configure --prefix=/usr/local/erlang
make && make install

# Add Erlang to PATH
export PATH=$PATH:/usr/local/erlang/bin
source /etc/profile

# 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
cd /usr/local/rabbitmq_server-3.8.16/sbin
./rabbitmq-server start

Verify the installation by opening http://localhost:15672/ in a browser.

Project setup (Java Spring Boot)

Add the AMQP starter dependency to pom.xml :

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

Configure RabbitMQ connection properties in application.yml :

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

Define a custom exchange, queue, and binding that use the delayed‑message plugin:

/**
 * 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 component that sets the x-delay header to specify the delay time:

@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;
            }
        );
    }
}

Receiver component that processes delayed messages:

@Component
@Slf4j
@RabbitListener(queues = "im.chat.cancel")
public class ChatQueueReceiver {
    @Autowired
    private ChatRestartRobotService chatRestartRobotService;

    @RabbitHandler
    public void handleOnChat(Long cmid) {
        chatRestartRobotService.restartRobot(cmid);
    }
}

After configuring the beans, simply invoke ChatQueueSender.sendMessageToChat(cmid, delay) where delay is the desired delay in milliseconds.

The article also includes practical analogies (flower shop scenario) to illustrate asynchronous processing, decoupling, and peak‑shaving benefits of a message queue.