抱歉,您的浏览器无法访问本站
本页面需要浏览器支持(启用)JavaScript
了解详情 >

RocketMQ延时消息

延迟消息

​ 延迟消息和定时消息,并无本质上的区别。延迟消息是指当前消息在创建之后的指定时间过后,再发送。定时消息,则是指消息在创建完成后,指定的时间发送。一般都是将消息存储在数据库中,使用定时任务进行消息扫描,满足条件的数据则发送给消费者。

延迟消息介绍

​ 基本概念:延迟消息是指生产者发送消息发送消息后,不能立刻被消费者消费,需要等待指定的时间后才可以被消费。

​ 场景案例:用户下了一个订单之后,需要在指定时间内(例如30分钟)进行支付,在到期之前可以发送一个消息提醒用户进行支付。

延时消息处理流程

Broker端内置延迟消息处理能力,核心实现思路都是一样:将延迟消息通过一个临时存储进行暂存,到期后才投递到目标Topic中。如下图所示:

延时消息处理流程

步骤说明如下

  1. producer要将一个延迟消息发送到某个Topic中
  2. Broker判断这是一个延迟消息后,将其通过临时存储进行暂存。
  3. Broker内部通过一个延迟服务(delay service)检查消息是否到期,将到期的消息投递到目标Topic中。这个的延迟服务名字为delay service,不同消息中间件的延迟服务模块名称可能不同。
  4. 消费者消费目标topic中的延迟投递的消息

显然,临时存储模块和延迟服务模块,是延迟消息实现的关键。上图中,临时存储和延迟服务都是在Broker内部实现,对业务透明。

​ 此外, 还有一些消息中间件原生并不支持延迟消息,如Kafka。在这种情况下,可以选择对Kafka进行改造,但是成本较大。另外一种方式是使用第三方临时存储,并加一层代理。

第三方存储选型要求

对于第三方临时存储,其需要满足以下几个特点:

高性能

​ 写入延迟要低,MQ的一个重要作用是削峰填谷,在选择临时存储时,写入性能必须要高,关系型数据库(如Mysql)通常不满足需求。

高可靠

​ 延迟消息写入后,不能丢失,需要进行持久化,并进行备份

支持排序

​ 支持按照某个字段对消息进行排序,对于延迟消息需要按照时间进行排序。普通消息通常先发送的会被先消费,延迟消息与普通消息不同,需要进行排序。例如先发一条延迟10s的消息,再发一条延迟5s的消息,那么后发送的消息需要被先消费。

支持长时间保存

​ 一些业务的延迟消息,需要延迟几个月,甚至更长,所以延迟消息必须能长时间保留。不过通常不建议延迟太长时间,存储成本比较大,且业务逻辑可能已经发生变化,已经不需要消费这些消息。

​ 例如,滴滴开源的消息中间件DDMQ,底层消息中间件的基础上加了一层代理,独立部署延迟服务模块,使用rocksdb进行临时存储。rocksdb是一个高性能的KV存储,并支持排序。

延时消息流转

此时对于延迟消息的流转如下图所示:

延迟消息的流转

说明如下

  1. 生产者将发送给producer proxy,proxy判断是延迟消息,将其投递到一个缓冲Topic中;

  2. delay service启动消费者,用于从缓冲topic中消费延迟消息,以时间为key,存储到rocksdb中;

  3. delay service判断消息到期后,将其投递到目标Topic中。

  4. 消费者消费目标topic中的数据

​ 这种方式的好处是,因为delay service的延迟投递能力是独立于broker实现的,不需要对broker做任何改造,对于任意MQ类型都可以提供支持延迟消息的能力,例如DDMQ对RocketMQ、Kafka都提供了秒级精度的延迟消息投递能力,但是Kafka本身并不支持延迟消息,而RocketMQ虽然支持延迟消息,但不支持秒级精度。

​ 事实上,DDMQ还提供了很多其他功能,仅仅从延迟消息的角度,完全没有必要使用这个proxy,直接将消息投递到缓冲Topic中,之后通过delay service完成延迟投递逻辑即可。

具体到delay service模块的实现上,也有一些重要的细节:

  1. 为了保证服务的高可用,delay service也是需要部署多个节点。

  2. 为了保证数据不丢失,每个delay service节点都需要消费缓冲Topic中的全量数据,保存到各自的持久化存储中,这样就有了多个备份,并需要以时间为key。不过因为是各自拉取,并不能保证强一致。如果一定要强一致,那么delay service就不需要内置存储实现,可以借助于其他支持强一致的存储。

  3. 为了避免重复投递,delay service需要进行选主,可以借助于zookeeper、etcd等实现。只有master可以通过生产者投递到目标Topic中,其他节点处于备用状态。否则,如果每个节点进行都投递,那么延迟消息就会被投递多次,造成消费重复。

  4. master要记录自己当前投递到的时间到一个共享存储中,如果master挂了,从slave节点中选出一个新的master节点,从之前记录时间继续开始投递。

  5. 延迟消息的取消:一些延迟消息在未到期之前,可能希望进行取消。通常取消逻辑实现较为复杂,且不够精确。对于那些已经快要到期的消息,可能还未取消之前,已经发送出去了,因此需要在消费者端做检查,才能万无一失。

RocketMQ中的延迟消息

开源RocketMQ支持延迟消息,但是不支持秒级精度。默认支持18个level的延迟消息,这是通过broker端的messageDelayLevel配置项确定的,如下:

1
messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

​ Broker在启动时,内部会创建一个内部主题:SCHEDULE_TOPIC_XXXX,根据延迟level的个数,创建对应数量的队列,也就是说18个level对应了18个队列。注意,这并不是说这个内部主题只会有18个队列,因为Broker通常是集群模式部署的,因此每个节点都有18个队列。

​ 延迟级别的值可以进行修改,以满足自己的业务需求,可以修改/添加新的level。例如:你想支持2天的延迟,修改最后一个level的值为2d,这个时候依然是18个level;也可以增加一个2d,这个时候总共就有19个level。

​ 可以看到这里并不支持秒级精度,按照《rocketmq developer guide》中的说法,是为了避免在broker对消息进行排序,造成性能影响。不过笔者考虑,之所以不支持,更多应该是商业上的考虑。

发送延迟消息

生产者在发送延迟消息非常简单,只需要设置一个延迟级别即可,注意不是具体的延迟时间,如:

1
2
3
4
5
6
7
Message msg=new Message();
msg.setTopic("topicTest");
msg.setTags("Tag");
msg.setBody("this is a delay message".getBytes());
//设置延迟level为5,对应延迟1分钟
msg.setDelayTimeLevel(5);
producer.send(msg);

如果设置的延迟level超过最大值,那么将会重置最最大值。

消费延时消费流程

延迟消息在RocketMQ Broker端的流转如下图所示:

延迟消息在RocketMQ Broker端的流转

可以看到,总共有6个步骤,下面会对这6个步骤进行详细的讲解:

  1. 修改消息Topic名称和队列信息

  2. 转发消息到延迟主题的CosumeQueue中

  3. 延迟服务消费SCHEDULE_TOPIC_XXXX消息

  4. 将信息重新存储到CommitLog中

  5. 将消息投递到目标Topic中

  6. 消费者消费目标topic中的数据

修改消息Topic名称和队列信息

​ RocketMQ Broker端在存储生产者写入的消息时,首先都会将其写入到CommitLog中。之后根据消息中的Topic信息和队列信息,将其转发到目标Topic的指定队列(ConsumeQueue)中。

​ 由于消息一旦存储到ConsumeQueue中,消费者就能消费到,而延迟消息不能被立即消费,所以这里将Topic的名称修改为SCHEDULE_TOPIC_XXXX,并根据延迟级别确定要投递到哪个队列下。

​ 同时,还会将消息原来要发送到的目标Topic和队列信息存储到消息的属性中。相关源码如下所示:org.apache.rocketmq.store.CommitLog#putMessage

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
public PutMessageResult putMessage(final MessageExtBrokerInner msg) {
...
// 如果是延迟消息
if (msg.getDelayTimeLevel() > 0) {
//如果设置的级别超过了最大级别,重置延迟级别
if (msg.getDelayTimeLevel() > this.defaultMessageStore.getScheduleMessageService()
.getMaxDelayLevel()) {
msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService()
.getMaxDelayLevel());
}

//修改Topic的投递目标为内部主题SCHEDULE_TOPIC_XXXX
topic = ScheduleMessageService.SCHEDULE_TOPIC;
//根据delayLevel,确定将消息投递到SCHEDULE_TOPIC_XXXX内部的哪个队列中
queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());

// 记录原始topic, queueId
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_TOPIC, msg.getTopic());
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_QUEUE_ID,
String.valueOf(msg.getQueueId()));
msg.setPropertiesString(MessageDecoder.messageProperties2String(msg.getProperties()));

//更新消息投递目标为SCHEDULE_TOPIC_XXXX和queueId
msg.setTopic(topic);
msg.setQueueId(queueId);
}
...
转发消息到延迟主题的CosumeQueue中

CommitLog中的消息转发到CosumeQueue中是异步进行的。在转发过程中,会对延迟消息进行特殊处理,主要是计算这条延迟消息需要在什么时候进行投递。

1
投递时间=消息存储时间(storeTimestamp) + 延迟级别对应的时间

​ 需要注意的是,会将计算出的投递时间当做消息Tag的哈希值存储到CosumeQueue中,CosumeQueue单个存储单元组成结构如下图所示:

CosumeQueue单个存储单元组成结构

参数介绍
  • Commit Log Offset:记录在CommitLog中的位置。
  • Size:记录消息的大小
  • Message Tag HashCode:记录消息Tag的哈希值,用于消息过滤。特别的,对于延迟消息,这个字段记录的是消息的投递时间戳。这也是为什么java中hashCode方法返回一个int型,只占用4个字节,而这里Message Tag HashCode字段确设计成8个字节的原因。
源码分析

相关源码参见:CommitLog#checkMessageAndReturnSize

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
public DispatchRequest checkMessageAndReturnSize(java.nio.ByteBuffer byteBuffer, final boolean checkCRC,
final boolean readBody) {
...
// Timing message processing
{
//如果消息需要投递到延迟主题SCHEDULE_TOPIC_XXX中
String t = propertiesMap.get(MessageConst.PROPERTY_DELAY_TIME_LEVEL);
if (ScheduleMessageService.SCHEDULE_TOPIC.equals(topic) && t != null) {
int delayLevel = Integer.parseInt(t);

if (delayLevel > this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) {
delayLevel = this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel();
}
//如果延迟级别大于0,计算目标投递时间,并将其当做tag哈希值
if (delayLevel > 0) {
tagsCode = this.defaultMessageStore.getScheduleMessageService()
.computeDeliverTimestamp(delayLevel,storeTimestamp);
}
}
}
...
延迟服务消费SCHEDULE_TOPIC_XXXX消息

Broker内部有一个ScheduleMessageService类,其充当延迟服务,消费SCHEDULE_TOPIC_XXXX中的消息,并投递到目标Topic中。

​ ScheduleMessageService在启动时,其会创建一个定时器Timer,并根据延迟级别的个数,启动对应数量的TimerTask,每个TimerTask负责一个延迟级别的消费与投递。

源码分析

相关源码如下所示:ScheduleMessageService#start

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
public void start() {
if (started.compareAndSet(false, true)) {
//1 创建定时器Timer
this.timer = new Timer("ScheduleMessageTimerThread", true);
//2 针对每个延迟级别,创建一个TimerTask
//2.1 迭代每个延迟级别:delayLevelTable是一个Map记录了每个延迟级别对应的延迟时间
for (Map.Entry<Integer, Long> entry : this.delayLevelTable.entrySet()) {
//2.2 获得每个每个延迟级别的level和对应的延迟时间
Integer level = entry.getKey();
Long timeDelay = entry.getValue();
Long offset = this.offsetTable.get(level);
if (null == offset) {
offset = 0L;
}
//2.3 针对每个级别创建一个对应的TimerTask
if (timeDelay != null) {
this.timer.schedule(new DeliverDelayedMessageTimerTask(level, offset),
FIRST_DELAY_TIME);
}
}
...

​ 需要注意的是,每个TimeTask在检查消息是否到期时,首先检查对应队列中尚未投递第一条消息,如果这条消息没到期,那么之后的消息都不会检查。如果到期了,则进行投递,并检查之后的消息是否到期。

将信息重新存储到CommitLog中

​ 在将消息到期后,需要投递到目标Topic。由于在第一步已经记录了原来的Topic和队列信息,因此这里重新设置,再存储到CommitLog即可。此外,由于之前Message Tag HashCode字段存储的是消息的投递时间,这里需要重新计算tag的哈希值后再存储。

源码参见:DeliverDelayedMessageTimerTask的messageTimeup方法。

将消息投递到目标Topic中

​ 这一步与第二步类似,不过由于消息的Topic名称已经改为了目标Topic。因此消息会直接投递到目标Topic的ConsumeQueue中,之后消费者即消费到这条消息。

延迟消息与消息重试的关系

​ RocketMQ提供了消息重试的能力,在并发模式消费的情况,如果消费失败,可以返回一个枚举值RECONSUME_LATER,那么消息之后将会进行重试。如:

1
2
3
4
5
6
7
8
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
ConsumeConcurrentlyContext context) {
//处理消息,失败,返回RECONSUME_LATER,进行重试
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
});
重试超时时常

重试默认会进行重试16次。使用过RocketMQ消息重试功能的用户,可能看到过以下这张图:

第几次重试 与上次重试的间隔时间 第几次重试 与上次重试的间隔时间
1 10 秒 9 7 分钟
2 30 秒 10 8 分钟
3 1 分钟 11 9 分钟
4 2 分钟 12 10 分钟
5 3 分钟 13 20 分钟
6 4 分钟 14 30 分钟
7 5 分钟 15 1 小时
8 6 分钟 16 2 小时

​ 细心地的读者发现了,消息重试的16个级别,实际上是把延迟消息18个级别的前两个level去掉了,事实上,RocketMQ的消息重试也是基于延迟消息来完成的。在消息消费失败的情况下,将其重新当做延迟消息投递回Broker,在投递回去时,会跳过前两个level,因此只重试16次。

代码案例

生产者

通过 msg.setDelayTimeLevel(5);代码设置延时一分钟

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
public class DelayedProducer {

public static void main(String[] args) throws Exception {
//创建一个消息生产者,并设置一个消息生产者组
DefaultMQProducer producer = new DefaultMQProducer("rocket_test_consumer_group");

//指定 NameServer 地址
producer.setNamesrvAddr("127.0.0.1:9876");

//初始化 Producer,整个应用生命周期内只需要初始化一次
producer.start();

//创建一条消息对象,指定其主题、标签和消息内容
Message msg = new Message(
/* 消息主题名 */
"topicTest",
/* 消息标签 */
"TagA",
/* 消息内容 */
("发送延时消息,当前时间" + System.currentTimeMillis()).getBytes(RemotingHelper.DEFAULT_CHARSET)
);
//设置延迟level为5,对应延迟1分钟
msg.setDelayTimeLevel(5);
//发送消息并返回结果
SendResult sendResult = producer.send(msg);

System.out.printf("%s%n", sendResult);

// 一旦生产者实例不再被使用则将其关闭,包括清理资源,关闭网络连接等
producer.shutdown();
}
}

消费者

消费者不需要做任何处理正常消费即可

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
public class DelaydConsumer {
public static void main(String[] args) throws Exception {
//创建一个消息消费者,并设置一个消息消费者组
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("rocket_test_consumer_group");
//指定 NameServer 地址
consumer.setNamesrvAddr("127.0.0.1:9876");
//设置 Consumer 第一次启动时从队列头部开始消费还是队列尾部开始消费
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
//订阅指定 Topic 下的所有消息
consumer.subscribe("topicTest", "*");

//注册消息监听器
consumer.registerMessageListener(new MessageListenerConcurrently() {
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
if (list != null) {
for (MessageExt ext : list) {
try {
String message = new String(ext.getBody(), RemotingHelper.DEFAULT_CHARSET);
System.out.println("Consumer-线程名称=[" + Thread.currentThread().getId() + "],接收时间:[" + System.currentTimeMillis() + "],消息=[" + message + "]");
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
}
}
// 业务方正常消费
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});

// 消费者对象在使用之前必须要调用 start 初始化
consumer.start();
System.out.println("消息消费者已启动");
}
}

测试

生产者发送消息后检查消费者接收到的消息

1
2
消息消费者已启动
Consumer-线程名称=[37],接收时间:[1608121246231],消息=[发送延时消息,消息发送时间1608121184151]

发送时间减去接收时间 可以看到 两个的差值约等于 60S 一分钟

评论