Implementing Delayed Messaging with Spring Cloud Stream and RabbitMQ

Application Scenario

When using open‑source schedulers such as elastic‑job, tasks often have fixed intervals (e.g., every 30 minutes or daily at 01:00). Some business scenarios require a start time that is only known later, like sending a blog post reminder two hours after publishing. Such dynamic scheduling can be achieved with Spring Cloud Stream and RabbitMQ delayed messages.

Hands‑on

Plugin Installation

RabbitMQ provides a delayed‑message plugin. Download the appropriate rabbitmq_delayed_message_exchange version from the official plugin page and place the .ez file into the plugins directory of your RabbitMQ installation.

Note: Supported only on RabbitMQ 3.6.x and above.

Enable the plugin without restarting:

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

If the plugin is not enabled, you may encounter errors like COMMAND_INVALID - unknown exchange type 'x-delayed-message'.

Application Code

The following example demonstrates a Spring Boot application that publishes a delayed message and consumes it after the specified delay.

@EnableBinding(TestApplication.TestTopic.class)
@SpringBootApplication
public class TestApplication {
    public static void main(String[] args) {
        SpringApplication.run(TestApplication.class, args);
    }

    @Slf4j
    @RestController
    static class TestController {
        @Autowired
        private TestTopic testTopic;

        @GetMapping("/sendMessage")
        public String messageWithMQ(@RequestParam String message) {
            log.info("Send: " + message);
            testTopic.output().send(MessageBuilder.withPayload(message)
                .setHeader("x-delay", 5000).build());
            return "ok";
        }
    }

    @Slf4j
    @Component
    static class TestListener {
        @StreamListener(TestTopic.INPUT)
        public void receive(String payload) {
            log.info("Received: " + payload);
        }
    }

    interface TestTopic {
        String OUTPUT = "example-topic-output";
        String INPUT = "example-topic-input";

        @Output(OUTPUT)
        MessageChannel output();

        @Input(INPUT)
        SubscribableChannel input();
    }
}

The controller defines the /sendMessage endpoint, adds an x-delay header (5 000 ms), and sends the message. The listener receives the message after the delay.

Producer Configuration

spring.cloud.stream.bindings.example-topic-output.destination=delay-topic
spring.cloud.stream.rabbit.bindings.example-topic-output.producer.delayed-exchange=true

The delayed-exchange flag tells Spring Cloud Stream to create an exchange with delayed‑message capabilities.

Consumer Configuration

spring.cloud.stream.bindings.example-topic-input.destination=delay-topic
spring.cloud.stream.bindings.example-topic-input.group=test
spring.cloud.stream.rabbit.bindings.example-topic-input.consumer.delayed-exchange=true

Without this flag, the consumer would fail with a PRECONDITION_FAILED error because the exchange type would be mismatched.

Running the Example

Start the application and invoke: http://localhost:8080/sendMessage?message=hello Log output shows a 5‑second gap between Send: hello and Received: hello, confirming the delayed delivery.

Deep Dive

To inspect delayed messages, open the RabbitMQ management UI, navigate to the Exchanges page, and locate the exchange of type x-delayed-message. Clicking the exchange reveals detailed statistics.

Code Repository

The full source code is available in the stream-delayed-message project on GitHub and Gitee.