延迟消息插件
使用 RabbitMQ 延迟消息交换器插件实现延迟消息投递。安排消息在未来某个时间处理，无需自定义计时器。
延迟消息插件 是 CoddyKit 上的免费 RabbitMQ Messaging & Async Systems 课时。 这是第 1 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 RabbitMQ Messaging & Async Systems 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 RabbitMQ Messaging & Async Systems 课程共包含 4 节课。
本课时的部分内容尚未翻译，以英文显示。
What are Delayed Messages?
Sometimes you don't want a message processed immediately. Think of sending a reminder email an hour from now, or processing a payment 30 minutes after an order is placed.
Delayed messages allow you to publish a message to RabbitMQ, but tell the broker to hold onto it for a specific duration before delivering it to consumers.
Scheduling Without the Plugin
Without a dedicated feature, scheduling messages can be complex:
- Custom Timers: You might build your own service with timers, but this adds complexity and a single point of failure.
- Polling Databases: Storing messages in a database and periodically checking for due times is inefficient.
- External Schedulers: Using CRON jobs or other external schedulers still requires your application to manage the message state.
RabbitMQ's Delayed Message Plugin simplifies this greatly!
Meet the `x-delayed-message` Plugin
The RabbitMQ Delayed Message Exchange plugin provides a special exchange type that can hold messages and release them after a specified delay.
It acts like a buffer, managing the delay internally without requiring your application to keep track of message timings.
This makes scheduling messages much simpler and more robust.
Delayed Exchange Mechanics
When you publish a message to an `x-delayed-message` exchange, you include a special header: x-delay. This header's value is the delay in milliseconds.
The exchange holds the message until its delay expires. Once expired, the message is routed to queues exactly as if it were a regular message published to an exchange of its original type (e.g., direct, fanout, topic).
Enabling the Plugin (Admin)
Before using it, the plugin must be enabled on your RabbitMQ server. This is typically done via the command line:
rabbitmq-plugins enable rabbitmq_delayed_message_exchangeA server restart might be required for the changes to take effect. Always ensure your broker has the plugin enabled before attempting to use it.
Declaring a Delayed Exchange
To use delayed messages, you first declare an exchange with the type x-delayed-message. You also specify its underlying type (e.g., direct, fanout, topic) which determines how messages are routed after the delay.
Here's how to declare a delayed direct exchange in Java:
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.util.HashMap;
import java.util.Map;
public class DeclareDelayedExchange {
private static final String EXCHANGE_NAME = "my_delayed_exchange";
public static void main(String[] argv) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost"); // Assuming RabbitMQ is local
try (Connection connection = factory.newConnection();
Channel channel = connection.createChannel()) {
Map<String, Object> args = new HashMap<>();
args.put("x-delayed-type", "direct"); // Underlying exchange type
channel.exchangeDeclare(EXCHANGE_NAME, "x-delayed-message", true, false, args);
System.out.println("Delayed exchange '" + EXCHANGE_NAME + "' declared.");
}
}
}Sending with `x-delay` Header
When publishing to your x-delayed-message exchange, you add an x-delay header to your message properties. The value is an integer representing the delay in milliseconds.
Let's send a message that will be delivered after 5 seconds (5000 milliseconds).
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.util.HashMap;
import java.util.Map;
public class DelayedMessageProducer {
private static final String EXCHANGE_NAME = "my_delayed_exchange";
private static final String ROUTING_KEY = "delayed.key";
public static void main(String[] argv) throws Exception {
ConnectionFactory factory = new ConnectionFactoryFactory();
factory.setHost("localhost");
try (Connection connection = factory.newConnection();
Channel channel = connection.createChannel()) {
// Ensure the delayed exchange is declared (from previous scene)
Map<String, Object> args = new HashMap<>();
args.put("x-delayed-type", "direct");
channel.exchangeDeclare(EXCHANGE_NAME, "x-delayed-message", true, false, args);
System.out.println("Exchange declared (if not exists).");
String message = "Hello, delayed world!";
int delayInMs = 5000; // 5 seconds
Map<String, Object> headers = new HashMap<>();
headers.put("x-delay", delayInMs); // Set the delay header
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
.headers(headers)
.build();
channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, props, message.getBytes("UTF-8"));
System.out.println(" [x] Sent '" + message + "' with delay " + delayInMs + "ms");
}
}
}Consuming Delayed Messages
Consumers don't need any special logic to receive delayed messages. Once the delay expires, the x-delayed-message exchange routes the message to the bound queues, and consumers receive it like any other message.
The key is that the message arrives at the consumer after the specified delay. You'll need to bind a queue to your delayed exchange with the appropriate routing key.
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import java.nio.charset.StandardCharsets;
import java.util.Date;
public class DelayedMessageConsumer {
private static final String EXCHANGE_NAME = "my_delayed_exchange";
private static final String QUEUE_NAME = "delayed_queue";
private static final String ROUTING_KEY = "delayed.key";
public static void main(String[] argv) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
// Declare the queue and bind it to the delayed exchange
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
System.out.println(" [x] Received '" + message + "' at " + new Date());
};
channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> {});
}
}Common Use Cases
Delayed messages are incredibly useful for:
- Reminders: Send a notification after a user's trial expires.
- Scheduled Tasks: Process a batch job at a specific time in the future.
- Retry Mechanisms: Requeue a failed message to be retried after a delay.
- Drip Campaigns: Send a series of emails over several days.
They remove the need for complex external scheduling services.
Quick Check: Delayed Message Basics
You've learned how to declare an x-delayed-message exchange and publish messages with a delay. Let's test your understanding.
Recap: Delayed Messages Plugin
In this lesson, you learned about the RabbitMQ Delayed Message Exchange plugin. It allows you to schedule messages to be delivered at a future time.
- You declare an exchange with type
x-delayed-messageand an underlying exchange type. - You publish messages with an
x-delayheader (value in milliseconds). - Consumers receive these messages normally, but only after the specified delay.
This powerful feature simplifies many time-based messaging patterns.
常见问题解答
「延迟消息插件」课时是免费的吗？
是的 — 「延迟消息插件」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 RabbitMQ Messaging & Async Systems 课程的其余内容，请升级到 CoddyKit PRO。 RabbitMQ Messaging & Async Systems 课程共包含 4 节课。
「延迟消息插件」这节课中我会学到什么？
使用 RabbitMQ 延迟消息交换器插件实现延迟消息投递。安排消息在未来某个时间处理，无需自定义计时器。 你通过在浏览器中直接运行的动手代码来练习 RabbitMQ Messaging & Async Systems，全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 RabbitMQ Messaging & Async Systems 需要有经验吗？
无需任何先前经验。CoddyKit 上的 RabbitMQ Messaging & Async Systems 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 1 节课，共 4 节。
「延迟消息插件」课时需要多长时间？
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。
我能在这节 RabbitMQ Messaging & Async Systems 课中编写并运行代码吗？
能。每节 RabbitMQ Messaging & Async Systems 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。