Schedule Messages in RabbitMQ using Dead-Letter Exchange & TTL — SpringAMQP

Photo by Artem Maltsev on Unsplash

There are some scenarios where we want to delay the delivery of certain messages so that it won’t be consumed immediately.
Few examples include:
1. Publishing a scheduled device software upgrade
2. Scheduling e-mails or push notification for events yet to occur

AMQP protocol doesn’t support delaying or scheduling messages. So here we use the Dead Letter Exchange & TTL message property as a work around.

Setting up RabbitMQ queues with dead letter exchange

Create an exchange say scheduled.dx and bind it to a queue scheduledq with routing key scheduled from which the delayed messages are to be consumed. This queue acts as our consumer queue.

Create a second exchange say delayed.dx to which the scheduled messages will be sent. Now create a queue delayedq with the following arguments x-dead-letter-exchange & x-dead-letter-routing-key.
The delayedq configuration should look like:

dead-letter configuration

Bind the delayedq to delayed.dx with routingKey say delay.

Posting delayed/scheduled messages via SpringAMQP

So, I have created a SpringBoot application to post scheduled messages to rabbitMq. If you want to jump to the code directly, please find it here: https://github.com/RasikFareed/schedule-message

We will be using Spring’s AMQP template to send & receive messages from RabbitMQ. Additionally, we need to write a CustomMessagePostProcessor to set an expiration(ttl) to the message.

public class CustomMessagePostProcessor implements MessagePostProcessor {
    private final Integer ttl;
    public CustomMessagePostProcessor(final Integer ttl) {
        this.ttl = ttl;
    }
    @Override
    public Message postProcessMessage(final Message message) throws AmqpException {
        message.getMessageProperties().setExpiration(ttl.toString());
        return message;
    }
}

Here you can see, the expiration is set to the message from the MessageProperties. So we could add this CustomMessagePostProcessor while we publish a message to rabbitMq, and the snippet to publish message with ttl should look like

rabbitTemplate.convertAndSend(exchange,
        routingKey,
        message,
        new CustomMessagePostProcessor(ttl)); 
        
//ttl is set in milliseconds

Running the application

Once the rabbitMq properties are configured in application.yaml, we are ready to start the application. On successful application start, use the following CURL to post scheduled message to rabbitMq

curl --location --request POST 'localhost:8080/schedule'  --header 'Content-Type: application/json'  --data-raw '{ "message":"hello world!", "ttl":30000 }'

As we can see in the console, there is a 30 second delay between the published & consumed message. Thus we are able to consume messages with delay.

Code & References:

RasikFareed/schedule-message

Spring AMQP

Delayed messages with RabbitMQ - Documentation - CloudAMQP