Delay message to send to listener using Spring AMQP

3
votes

I have a requirement to send message to MessageListener after certain duration , So is there any way to achieve using Spring AMQP.

Eg . Producer produces the message and message goes to RabbitMQ Q , The message gets received Listener listening to that Q immediately, I want to delay that message to be received at consumer side say after some configuration parameter say 1000ms

2
rabbitmqspring-amqpspring-rabbit

2 Answers

6
votes

The RabbitMQ provides for this purpose Delayed Exchange feature.

Starting with version 1.6 Spring AMQP also provides a high level API on the matter: http://docs.spring.io/spring-amqp/reference/html/_reference.html#delayed-message-exchange:

<rabbit:topic-exchange name="topic" delayed="true" />

MessageProperties properties = new MessageProperties();
properties.setDelay(15000);
template.send(exchange, routingKey,
        MessageBuilder.withBody("foo".getBytes()).andProperties(properties).build());

UPDATE

Before Spring AMQP 1.6 you should do like this:

@Bean
CustomExchange delayExchange() {
    Map<String, Object> args = new HashMap<String, Object>();
    args.put("x-delayed-type", "direct");
    return new CustomExchange("my-exchange", "x-delayed-message", true, false, args);
}

...

MessageProperties properties = new MessageProperties();
properties.setHeader("x-delay", 15000);
template.send(exchange, routingKey,
        MessageBuilder.withBody("foo".getBytes()).andProperties(properties).build());

Also see this question and its answer: Scheduled/Delay messaging in Spring AMQP RabbitMq

3
votes

If you use spring boot, it can be like this:

@Bean
Queue queue() {
    return QueueBuilder.durable(queueName)
            .withArgument("x-dead-letter-exchange", dlx)
            .withArgument("x-dead-letter-routing-key", dlq)
            .build();
}


@Bean
TopicExchange exchange() {
    return (TopicExchange) ExchangeBuilder.topicExchange(topicExchangeName)
            .delayed()
            .build();

@Bean
Binding binding() {
    return BindingBuilder.bind(queue()).to(exchange()).with(queueName);
}