延时消息
定时消息是 Apache RocketMQ 中的一种高级特性消息。本文介绍定时消息与延时消息的应用场景、工作机制、使用限制、使用示例以及使用注意事项。
定时消息和延时消息本质上是相同的，它们都是根据消息设定的时间，在固定的时间点将消息投递给消费者。因此，下文统一使用“定时消息”这一概念。
适用场景
在分布式定时调度和任务超时处理等场景中，需要精确且可靠的基于时间的事件触发。Apache RocketMQ 提供的定时消息功能，可以简化定时调度任务的开发，实现高性能、可扩展、可靠的定时触发。
场景一：分布式定时调度 
分布式定时调度场景涉及各种时间粒度的任务，例如每天 5 点执行文件清理的任务，或每 2 分钟触发一次推送消息的任务。传统的基于数据库数据集的定时调度方案在分布式场景下复杂且低效。相比之下，Apache RocketMQ 的定时消息允许您封装多种类型的时间触发器。
场景二：任务超时处理 
任务超时处理的一个典型场景是电商支付，即未支付的订单在经过一段特定时间后会被取消，而不是立即取消。在这种情况下，您可以使用 Apache RocketMQ 的定时消息来检查并触发超时任务。
基于定时消息的任务超时处理具有以下优势：
多种时间粒度且开发简便：Apache RocketMQ 的定时消息不受固定时间增量的限制。您可以按任意时间粒度触发任务，且无需去重。
高性能和可扩展性：Apache RocketMQ 的定时消息具有高并发和可扩展性。这优于传统的数据库扫描方法，因为数据库扫描实现复杂，且频繁的 API 调用扫描可能会导致性能瓶颈。
工作原理
定时消息定义
定时消息是 Apache RocketMQ 中的一种高级特性消息。定时消息允许消费者在消息发送到服务端后，仅在指定的时间点或经过指定的时间段后才能消费。您可以使用定时消息在分布式场景中实现延迟调度和触发。
时间设置规则
Apache RocketMQ 中定时消息的定时或延时时间以时间戳表示，而非时间周期。
定时时间为毫秒级的 Unix 时间戳。您必须将消息投递的计划时间转换为毫秒级的 Unix 时间戳。您可以使用 Unix 时间戳转换器将时间转换为毫秒级 Unix 时间戳。
定时时间必须在允许的时间范围内。如果定时时间超出该范围，定时设置将不会生效，消息会立即在服务端被投递。
默认情况下，定时消息的最大定时时长为 24 小时。该默认值不可更改。更多信息，请参见参数限制。
定时时间必须晚于当前时间。如果定时时间设置为早于当前时间，定时设置将不会生效，消息会立即在服务端被投递。
以下是两个时间设置示例：
定时消息：如果当前时间为 2022-06-09 17:30:00，您希望在 2022-06-09 19:20:00 投递消息，则定时时间的毫秒级 Unix 时间戳为 1654773600000。
延时消息：如果当前时间为 2022-06-09 17:30:00，您希望在 1 小时后投递消息，则消息投递时间为 2022-06-09 18:30:00，对应的毫秒级 Unix 时间戳为 1654770600000。
定时消息生命周期
初始化：消息由生产者构建并初始化，准备发送到服务端。
定时：消息发送到服务端后，存储在基于时间的存储系统中，直到指定的投递时间。此时不会立即为该消息创建索引。
就绪：在指定时间点，消息被写入常规存储引擎，此时消息对消费者可见，并等待消费者消费。
处理中（Inflight）：消息被消费者获取，并按照本地业务逻辑进行处理。
在此期间，Broker 等待消费者提交结果。若超时未收到反馈，Apache RocketMQ 会对该消息进行重试。详情请参考消费重试。
确认（Acked）：消费者完成消费并提交结果，Broker 标记消息消费状态。
默认情况下，Apache RocketMQ 保留所有消息。消费确认后仅进行逻辑标记，不会立即物理删除。因此，在消息过期或空间不足被清理前，消费者可以回溯消费。
- 删除：当消息过期或存储空间不足时，Apache RocketMQ 会按滚动方式删除最早的物理文件。详情请参考消息存储与清理。
使用限制
消息类型一致性
定时消息只能发送到 MessageType 为 Delay 的主题（Topic）中。
时间粒度
Apache RocketMQ 定时消息的时间粒度精确到毫秒。默认粒度值为 1000 毫秒。
Apache RocketMQ 定时消息的状态可以持久化存储。如果消息系统发生故障并重启，消息仍会根据指定的投递时间进行投递。然而，如果存储系统发生异常或重启，定时消息的投递可能会出现延迟。
示例
创建主题（Topic）
在 Apache RocketMQ 5.0 中，建议使用 mqadmin 工具创建主题。注意，需要添加消息类型作为属性参数。以下是示例：
sh mqadmin updateTopic -n <nameserver_address> -t <topic_name> -c <cluster_name> -a +message.type=Delay
发送消息
与普通消息不同，定时消息必须为它们指定一个投递时间戳。
创建 DELAY 主题
/bin/mqadmin updateTopic -c DefaultCluster -t DelayTopic -n 127.0.0.1:9876 -a +message.type=DELAY
- -c 集群名称
- -t 主题名称
- -n nameserver 地址
- -a 额外属性，我们添加一个
message.type属性，并将值设为DELAY，以支持发送 DELAY 消息。
以下代码提供了发送和消费定时消息的 Java 示例：
// Send delay messages.
MessageBuilder messageBuilder = null;
// Specify a millisecond-level Unix timestamp. In this example, the specified timestamp indicates that the message will be delivered in 10 minutes from the current time.
Long deliverTimeStamp = System.currentTimeMillis() + 10L * 60 * 1000;
Message message = messageBuilder.setTopic("topic")
// Specify the message index key. The system uses the key to locate the message.
.setKeys("messageKey")
// Specify the message tag. The consumer can use the tag to filter messages.
.setTag("messageTag")
.setDeliveryTimestamp(deliverTimeStamp)
// Configure the message body.
.setBody("messageBody".getBytes())
.build();
try {
// Send the messages. Focus on the result of message sending and exceptions such as failures.
SendReceipt sendReceipt = producer.send(message);
System.out.println(sendReceipt.getMessageId());
} catch (ClientException e) {
e.printStackTrace();
}
// Consumption example 1: If a scheduled message is consumed by a push consumer, the consumer needs to process the message only in the message listener.
MessageListener messageListener = new MessageListener() {
@Override
public ConsumeResult consume(MessageView messageView) {
System.out.println(messageView.getDeliveryTimestamp());
// Return the status based on the consumption result.
return ConsumeResult.SUCCESS;
}
};
// Consumption example 2: If a scheduled message is consumed by a simple consumer, the consumer must obtain the message for consumption and submit the consumption result.
List<MessageView> messageViewList = null;
try {
messageViewList = simpleConsumer.receive(10, Duration.ofSeconds(30));
messageViewList.forEach(messageView -> {
System.out.println(messageView);
// After consumption is complete, the consumer must invoke ACK to submit the consumption result.
try {
simpleConsumer.ack(messageView);
} catch (ClientException e) {
e.printStackTrace();
}
});
} catch (ClientException e) {
// If the pull fails due to system traffic throttling or other reasons, you must re-initiate the request to obtain the message.
e.printStackTrace();
}
}
使用说明
我们建议您不要为大量消息设置相同的投递时间。
定时消息在被投递给消费者之前，会存储在基于时间的存储系统中。如果您为大量定时消息指定了相同的投递时间，系统必须在到达该时间点时同时处理这些消息。这会使系统负载过重，并导致消息投递出现延迟。