RocketMQ 8：延时消息

1.Producer 将消息发送到消息队列 RocketMQ 服务端，但并不期望这条消息立马投递，而是延迟一定时间后才投递到 Consumer 进行消费， 该消息即延时消息。

适用场景： 消息生产和消费有时间窗口要求：比如在电商交易中超时未支付关闭订单的场景，在订单创建时会发送一条延时消息。这条消息将会在 30 分钟以 后投递给消费者，消费者收到此消息后需要判断对应的订单是否已完成支付。 如支付未完成，则关闭订单。如已完成支付则忽略。 "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"是18个等级，level为1，表示延迟1秒后消费，level为5表示延迟1分钟后消费，level为18表示延迟2个小时消费。生产消息跟普通的生产消息类似，只需要在消息上设置延迟队列的 level 即可。

2.创建生产者TimerProducer，使用setDelayTimeLevel方法设置等级，这里设置3，延迟10s

package org.example.timer; import org.apache.rocketmq.client.producer.DefaultMQProducer; import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.remoting.common.RemotingHelper; /** * 延时发送 */ public class TimerProducer { public static void main(String[] args) throws Exception { //生产者实例化 DefaultMQProducer producer = new DefaultMQProducer("sync"); //指定mq服务器地址 producer.setNamesrvAddr("192.168.42.112:9876"); //启动实例 producer.start(); for (int i = 0; i < 10; i++) { //创建一个消息实例，指定topic、tag和消息体 Message msg = new Message("TopicA", "TagB", ("Hello RocketMQ" + i).getBytes(RemotingHelper.DEFAULT_CHARSET)); //这个是设置延时消息的属性 //"1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h" 18个等级 msg.setDelayTimeLevel(3); //发送消息 SendResult sendResult = producer.send(msg); System.out.println("status:" + sendResult.getSendStatus()+";MsgId:" + sendResult.getMsgId() +";QueueId:" + sendResult.getMessageQueue().getQueueId()+";value:" + new String(msg.getBody())); } //实例关闭 producer.shutdown(); } }

3.创建消费者TimerConsumer

package org.example.timer; import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext; import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; import org.apache.rocketmq.client.exception.MQClientException; import org.apache.rocketmq.common.consumer.ConsumeFromWhere; import org.apache.rocketmq.common.message.MessageExt; import java.util.List; /** * 消费者-推模式 **/ public class TimerConsumer { public static void main(String[] args) throws MQClientException { DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("timeGroup"); consumer.subscribe("TopicA", "*"); consumer.setNamesrvAddr("192.168.42.112:9876"); consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);//每次从最后一次消费的偏移量开始消费 consumer.registerMessageListener(new MessageListenerConcurrently() { public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) { System.out.println("queueID:" + list.get(0).getQueueId() + ",ThreadName:" + Thread.currentThread().getName() + ",Messages:" + new String(list.get(0).getBody())); return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } }); consumer.start(); System.out.println("ConsumerPartOrder Started."); } }

4.启动消费者，再启动生产者

5.查看消费者打印，消息经过10s之后，才被消费者消费