大家好，我是君哥。今天来聊一聊 RocketMQ 的延时消息是怎么实现的。,延时消息是指发送到 RocketMQ 后不会马上被消费者拉取到，而是等待固定的时间，才能被消费者拉取到。,延时消息的使用场景很多，比如电商场景下关闭超时未支付的订单，某些场景下需要在固定时间后发送提示消息。,首先看一个生产者发送延时消息的官方示例代码：,从上面的代码可以看到，跟普通消息不一样的是，消息设置 setDelayTimeLevel 属性值，这里设置为 3，这里最终将 3 这个延时级别复制给了 DELAY 属性。,关于延时级别，可以看下面这个定义：,这里延时级别有 18 个，上面的示例代码中延迟级别是 3，消息会延迟 10s 后消费者才能拉取。,Broker 收到消息后，会将消息写入 CommitLog。在写入时，会判断消息 DELAY 属性是否大于 0。代码如下：,从上面的代码可以看到，CommitLog 写入时并没有直接写入，而是把 Topic 改为 SCHEDULE_TOPIC_XXXX，把 queueId 改为延时级别减 1。因为延时级别有 18 个，所以这里有 18 个队列。如下图：,延时消息写入后，会有一个调度任务不停地拉取这些延时消息，这个逻辑在类 ScheduleMessageService。这个类的初始化代码如下：,上面的 load() 方法会加载一个 delayLevelTable（ConcurrentHashMap类型），key 保存延时级别（从 1 开始），value 保存延时时间(单位是 ms)。,load() 方法结束后，创建了一个有 18 个核心线程的定时线程池，然后遍历 delayLevelTable，创建 18 个任务（DeliverDelayedMessageTimerTask）进行每个延时级别的任务调度。任务调度的代码逻辑如下：,这段代码可以参考下面的流程图来进行理解：,上面有一个修正投递时间的函数，这个函数的意义是如果已经过了投递时间，那么立即投递。代码如下：,注意：消息从 CommitLog 转发到 ConsumeQueue 时，会判断是否是延时消息（Topic = SCHEDULE_TOPIC_XXXX 并且延时级别大于 0），如果是延时消息，就会修改 tagsCode 值为消息投递的时间戳，而 tagsCode 原值是 tag 的 HashCode。代码如下：,如下图：,而 ScheduleMessageService 调度线程将消息从 ConsumeQueue 重新投递到原始队列中时，会把 tagsCode 再次修改为 tag 的 HashCode，代码如下：,如下图：,如果有一个业务场景，要求延时消息 3 小时才能消费，而 RocketMQ 的延时消息最大延时级别只支持延时 2 小时，怎么处理？,这里提供两个思路供大家参考：,在 Broker 上修改 messageDelayLevel 的默认配置；,在客户端缓存 msgId，先设置延时级别是 18（2h），当客户端拉取到消息后首先判断有没有缓存，如果有缓存则再次发送延时消息，这次延时级别是 17（1h），如果没有缓存则进行消费。,经过上面的讲解，延时消息的处理流程如下：,最后，延时消息的延时时间并不精确，这个时间是 Broker 调度线程把消息重新投递到原始的 MessageQueue 的时间，如果发生消息积压或者 RocketMQ 客户端发生流量管控，客户端拉取到消息后进行处理的时间可能会超出预设的延时时间。
© 版权声明
文章版权归作者所有，未经允许请勿转载。
">