RocketMQ第五章：手把手教老婆实现-延时消息18禁

末蓝、 2022-11-18 01:40 237阅读 0赞

技术活，该赏
    关注+一键三连（点赞，评论，收藏）再看，养成好习惯

[RocketMQ使用教程相关系列 目录][RocketMQ_]

--------------------

**目录**

第一节：介绍

第二节：延时消息-生产者和消息者步骤说明

延时消息生产者代码实现步骤

延时消息消费者代码实现步骤

第三节：延时消息生产者

效果：

第四节：延时消息消费者

效果：

--------------------

# 第一节：介绍 #

比如电商里，提交了一个订单就可以发送一个延时消息，1h后去检查这个订单的状态，如果还是未付款就取消订单释放库存。

它的实现和普通消息的生产者，消费者基本一样，多了一个设置延迟级别。

message.setDelayTimeLevel()

现在RocketMq（开源的版本，白嫖的，知足吧）并不支持任意时间的延时，需要设置几个固定的延时等级，从1s到2h分别对应着等级1到18

注：阿里云收费版本支持任意时间延时

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

# 第二节：延时消息-生产者和消息者步骤说明 #

## 延时消息生产者代码实现步骤 ##

1.创建消息生产者producer，并制定生产者组名

2.指定Nameserver地址

3.启动producer

4.创建消息对象，指定主题Topic、Tag和消息体,设置延时级别

5.发送消息

6.关闭生产者producer

## 延时消息消费者代码实现步骤 ##

1.创建消费者Consumer，制定消费者组名

2.指定Nameserver地址

3.订阅主题Topic和Tag

4.设置回调函数，处理消息

5.启动消费者consumer

注意：消费者的 Topic 和 Tag 需要和生产者保持一致

# 第三节：延时消息生产者 #

public class Producer {
    
       public static void main(String[] args)
             throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
          // 1.创建消息生产者producer，并制定生产者组名
          DefaultMQProducer producer = new DefaultMQProducer("demo_producer_delay_group");
          // 2.指定Nameserver地址
          producer.setNamesrvAddr("192.168.88.131:9876");
          // 3.启动producer
          producer.start();
          System.out.println("生产者启动");
    
          for (int i = 0; i < 10; i++) {
             // 4.创建消息对象，指定主题Topic、Tag和消息体
             /**
              * 参数一：消息主题Topic
              * 参数二：消息Tag
              * 参数三：消息内容
              */
             Message msg = new Message("DelayTopic", "Tag1", ("Hello 虚竹" + i).getBytes());
             // 设定延迟时间 10s
             msg.setDelayTimeLevel(3);
             // 5.发送消息
             SendResult result = producer.send(msg);
             // 发送状态
             SendStatus status = result.getSendStatus();
    
             System.out.println("发送结果:" + result);
    
             // 线程睡1秒
             TimeUnit.SECONDS.sleep(1);
          }
    
          // 6.关闭生产者producer
          producer.shutdown();
       }
    
    }

## 效果： ##

![0][]

# 第四节：延时消息消费者 #

public class Consumer {
    
       public static void main(String[] args) throws Exception {
          // 1.创建消费者Consumer，制定消费者组名
          DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("demo_producer_delay_group");
          // 2.指定Nameserver地址
          consumer.setNamesrvAddr("192.168.88.131:9876");
          // 3.订阅主题Topic和Tag
          consumer.subscribe("DelayTopic", "*");
    
          // 4.设置回调函数，处理消息
          consumer.registerMessageListener(new MessageListenerConcurrently() {
    
             // 接受消息内容
             @Override
             public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                   System.out.println("消息ID：【" + msg.getMsgId() + "】,消息内容：" + new String(msg.getBody()) + ",延迟时间："
                         + (System.currentTimeMillis() - msg.getBornTimestamp()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
             }
          });
          // 5.启动消费者consumer
          consumer.start();
    
          System.out.println("消费者启动");
       }
    }

## 效果： ##

在10s后收到延时消息

![0][0 1]

[RocketMQ_]: https://blog.csdn.net/shi_hong_fei_hei/article/details/115470425
[0]: /images/20221022/e9b05723acb4486589ed38da90410802.png
[0 1]: /images/20221022/5ae9daf5288040af9f2ffb5a04969881.png