RocketMQ 延迟队列

深藏阁楼爱情的钟 2022-09-04 14:46 200阅读 0赞

# RocketMQ 延迟队列 #

## 什么是延迟队列 ##

指消息发送到某个队列后，在指定多长时间之后才能被消费。

## 应用场景 ##

## RocketMQ 延迟队列 ##

定时消息（延迟队列）是指消息发送到broker后，不会立即被消费，等待特定时间投递给真正的topic。

broker有配置项messageDelayLevel，默认值为“1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m  
8m 9m 10m 20m 30m 1h 2h”，18个level。

可以配置自定义messageDelayLevel。需要注意的是 messageDelayLevel是broker的属性，不属于某个topic。发消息时，设置delayLevel等级即可：

msg.setDelayLevel(level)

level有以下三种情况：

level == 0，消息为非延迟消息1<=level<=maxLevel，消息延迟特定时间，例如level1，延迟1s  
level > maxLevel，则level maxLevel，例如level==20，延迟2h

在 RocketMQ中定时消息会暂存在名为`SCHEDULE_TOPIC_XXXX`的topic中，并根据delayTimeLevel存入特定的queue，queueId = delayTimeLevel – 1，即一个queue只存相同延迟的消息，保证具有相同发送延迟的消息能够顺序消费。broker会调度地消费`SCHEDULE_TOPIC_XXXX`，将消息写入真实的topic。需要注意的是，定时消息会在第一次写入和调度写入真实topic时都会计数，因此发送数量、tps都会变高。

###### RocketMQ 延迟队列和RabbitMQ延迟队列相比 ######

RocketMQ直接一步到位，功能类似于`RabbitMQ的延迟交换器插件`、而RabbitMQ提供了延迟队列，但如果是基于消息设置每个消息设置不同的延迟时间，`会产生前面的消息早已过期，但后面的消息还存在消息队列中`，故此RabbitMQ提供了`延迟交换器插件`，而RocketMQ 延迟队列设计就比较好了。

##### 具体示例 #####

生产者

// 实例化生产者，并指定生产组名称
            DefaultMQProducer producer = new DefaultMQProducer("myproducer_group_topic_name_delay_01");
            //设置实例名称，一个jvm中有多个生产者可以根据实例名区分
            //默认default
            producer.setInstanceName("topic_delay");
            // 指定nameserver的地址
            producer.setNamesrvAddr("localhost:9876");
            //设置同步重试次数
            producer.setRetryTimesWhenSendFailed(2);
            //设置异步发送次数
            //producer.setRetryTimesWhenSendAsyncFailed(2);
            // 初始化生产者
            producer.start();
            for (int i = 0; i <20 ; i++) { 
                Message message = new Message("topic_name_delay", ("key=" + i).getBytes("utf-8"));
                //设置延迟消费时间 设置延迟时间级别0,18,0表示不延迟，18表示延迟2h，大于18的都是2h
                message.setDelayTimeLevel(i);
                // 1 同步发送 如果发送失败会根据重试次数重试
                SendResult send = producer.send(message);
                SendStatus sendStatus = send.getSendStatus();
                System.out.println(sendStatus.toString());
    
            }

消费者

/** * 推消息消费 */
            DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer("consumer_group_delay_01");
            // 指定nameserver的地址
            defaultMQPushConsumer.setNamesrvAddr("localhost:9876");
            defaultMQPushConsumer.subscribe("topic_name_delay", "*");
            // 1 提高消费并行度
            defaultMQPushConsumer.setConsumeThreadMax(10);
            defaultMQPushConsumer.setConsumeThreadMin(1);
    
            // 2 以批量方式进行 消费
            // 设置消息批处理的一个批次中消息的最大个数
            defaultMQPushConsumer.setConsumeMessageBatchMaxSize(10);
            //设置重试次数 默认16次
            defaultMQPushConsumer.setMaxReconsumeTimes(1);
            // 添加消息监听器，一旦有消息推送过来，就进行消费
            defaultMQPushConsumer.setMessageListener(new MessageListenerConcurrently() { 
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { 
                    //final MessageQueue messageQueue = context.getMessageQueue();
                    for (MessageExt msg : msgs) { 
                        System.out.println(msg);
                        try { 
                            System.out.println(new String(msg.getBody(), "utf-8"));
                        } catch (UnsupportedEncodingException e) { 
                            e.printStackTrace();
                        }
                    }
                    // 消息消费成功
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    // 消息消费失败
    // return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            });

消费消息，按照0到18级别来，0 表示不延迟，1表示延迟1s，大于等于18表示延迟2h  
按照级别一次类推  
`默认值为“1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m8m 9m 10m 20m 30m 1h 2h”，18个level`。

这里只拷贝0-8 的打印日志，可以自己等待确认。

> MessageExt \[queueId=0, storeSize=195, queueOffset=19, sysFlag=0, bornTimestamp=1628949643548, bornHost=/192.168.0.103:55518, storeTimestamp=1628949643554, storeHost=/192.168.0.103:10911, msgId=C0A8006700002A9F0000000000029A8A, commitLogOffset=170634, bodyCRC=858365373, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message\{topic=‘topic\_name\_delay’, flag=0, properties=\{MIN\_OFFSET=0, MAX\_OFFSET=20, CONSUME\_START\_TIME=1628949650433, UNIQ\_KEY=C0A8006748D07C53A9EB47ABD51C0000, CLUSTER=DefaultCluster, WAIT=true, DELAY=0\}, body=\[107, 101, 121, 61, 48\], transactionId=‘null’\}\]  
> key=0  
> MessageExt \[queueId=1, storeSize=234, queueOffset=22, sysFlag=0, bornTimestamp=1628949643558, bornHost=/192.168.0.103:55518, storeTimestamp=1628949644566, storeHost=/192.168.0.103:10911, msgId=C0A8006700002A9F000000000002ACF8, commitLogOffset=175352, bodyCRC=1143909675, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message\{topic=‘topic\_name\_delay’, flag=0, properties=\{MIN\_OFFSET=0, REAL\_TOPIC=topic\_name\_delay, MAX\_OFFSET=23, CONSUME\_START\_TIME=1628949650435, UNIQ\_KEY=C0A8006748D07C53A9EB47ABD5260001, CLUSTER=DefaultCluster, WAIT=true, DELAY=1, REAL\_QID=1\}, body=\[107, 101, 121, 61, 49\], transactionId=‘null’\}\]  
> key=1  
> MessageExt \[queueId=2, storeSize=234, queueOffset=18, sysFlag=0, bornTimestamp=1628949643561, bornHost=/192.168.0.103:55518, storeTimestamp=1628949648566, storeHost=/192.168.0.103:10911, msgId=C0A8006700002A9F000000000002ADE2, commitLogOffset=175586, bodyCRC=1562901649, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message\{topic=‘topic\_name\_delay’, flag=0, properties=\{MIN\_OFFSET=0, REAL\_TOPIC=topic\_name\_delay, MAX\_OFFSET=19, CONSUME\_START\_TIME=1628949650436, UNIQ\_KEY=C0A8006748D07C53A9EB47ABD5290002, CLUSTER=DefaultCluster, WAIT=true, DELAY=2, REAL\_QID=2\}, body=\[107, 101, 121, 61, 50\], transactionId=‘null’\}\]  
> key=2  
> MessageExt \[queueId=3, storeSize=234, queueOffset=17, sysFlag=0, bornTimestamp=1628949643566, bornHost=/192.168.0.103:55518, storeTimestamp=1628949653569, storeHost=/192.168.0.103:10911, msgId=C0A8006700002A9F000000000002AECC, commitLogOffset=175820, bodyCRC=706792455, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message\{topic=‘topic\_name\_delay’, flag=0, properties=\{MIN\_OFFSET=0, REAL\_TOPIC=topic\_name\_delay, MAX\_OFFSET=18, CONSUME\_START\_TIME=1628949653572, UNIQ\_KEY=C0A8006748D07C53A9EB47ABD52E0003, CLUSTER=DefaultCluster, WAIT=true, DELAY=3, REAL\_QID=3\}, body=\[107, 101, 121, 61, 51\], transactionId=‘null’\}\]  
> key=3  
> MessageExt \[queueId=0, storeSize=234, queueOffset=20, sysFlag=0, bornTimestamp=1628949643568, bornHost=/192.168.0.103:55518, storeTimestamp=1628949673574, storeHost=/192.168.0.103:10911, msgId=C0A8006700002A9F000000000002B0A0, commitLogOffset=176288, bodyCRC=876894628, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message\{topic=‘topic\_name\_delay’, flag=0, properties=\{MIN\_OFFSET=0, REAL\_TOPIC=topic\_name\_delay, MAX\_OFFSET=21, CONSUME\_START\_TIME=1628949673577, UNIQ\_KEY=C0A8006748D07C53A9EB47ABD5300004, CLUSTER=DefaultCluster, WAIT=true, DELAY=4, REAL\_QID=0\}, body=\[107, 101, 121, 61, 52\], transactionId=‘null’\}\]  
> key=4  
> MessageExt \[queueId=1, storeSize=234, queueOffset=23, sysFlag=0, bornTimestamp=1628949643570, bornHost=/192.168.0.103:55518, storeTimestamp=1628949703574, storeHost=/192.168.0.103:10911, msgId=C0A8006700002A9F000000000002B18A, commitLogOffset=176522, bodyCRC=1128491314, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message\{topic=‘topic\_name\_delay’, flag=0, properties=\{MIN\_OFFSET=0, REAL\_TOPIC=topic\_name\_delay, MAX\_OFFSET=24, CONSUME\_START\_TIME=1628949703577, UNIQ\_KEY=C0A8006748D07C53A9EB47ABD5320005, CLUSTER=DefaultCluster, WAIT=true, DELAY=5, REAL\_QID=1\}, body=\[107, 101, 121, 61, 53\], transactionId=‘null’\}\]  
> key=5  
> MessageExt \[queueId=2, storeSize=234, queueOffset=20, sysFlag=0, bornTimestamp=1628949643572, bornHost=/192.168.0.103:55518, storeTimestamp=1628949763575, storeHost=/192.168.0.103:10911, msgId=C0A8006700002A9F000000000002B35E, commitLogOffset=176990, bodyCRC=1514813576, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message\{topic=‘topic\_name\_delay’, flag=0, properties=\{MIN\_OFFSET=0, REAL\_TOPIC=topic\_name\_delay, MAX\_OFFSET=21, CONSUME\_START\_TIME=1628949763580, UNIQ\_KEY=C0A8006748D07C53A9EB47ABD5340006, CLUSTER=DefaultCluster, WAIT=true, DELAY=6, REAL\_QID=2\}, body=\[107, 101, 121, 61, 54\], transactionId=‘null’\}\]  
> key=6  
> MessageExt \[queueId=3, storeSize=234, queueOffset=19, sysFlag=0, bornTimestamp=1628949643574, bornHost=/192.168.0.103:55518, storeTimestamp=1628949823580, storeHost=/192.168.0.103:10911, msgId=C0A8006700002A9F000000000002B532, commitLogOffset=177458, bodyCRC=760023070, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message\{topic=‘topic\_name\_delay’, flag=0, properties=\{MIN\_OFFSET=0, REAL\_TOPIC=topic\_name\_delay, MAX\_OFFSET=20, CONSUME\_START\_TIME=1628949823583, UNIQ\_KEY=C0A8006748D07C53A9EB47ABD5360007, CLUSTER=DefaultCluster, WAIT=true, DELAY=7, REAL\_QID=3\}, body=\[107, 101, 121, 61, 55\], transactionId=‘null’\}\]  
> key=7  
> MessageExt \[queueId=0, storeSize=234, queueOffset=22, sysFlag=0, bornTimestamp=1628949643576, bornHost=/192.168.0.103:55518, storeTimestamp=1628949883582, storeHost=/192.168.0.103:10911, msgId=C0A8006700002A9F000000000002B706, commitLogOffset=177926, bodyCRC=1039275407, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message\{topic=‘topic\_name\_delay’, flag=0, properties=\{MIN\_OFFSET=0, REAL\_TOPIC=topic\_name\_delay, MAX\_OFFSET=23, CONSUME\_START\_TIME=1628949883586, UNIQ\_KEY=C0A8006748D07C53A9EB47ABD5380008, CLUSTER=DefaultCluster, WAIT=true, DELAY=8, REAL\_QID=0\}, body=\[107, 101, 121, 61, 56\], transactionId=‘null’\}\]  
> key=8

同样我们在控制台可以看到，存放的消息![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70]: /images/20220829/156406875a9b4583a80a7b2ff73b3c18.png