RocketMQ延时消息使用

╰半夏微凉° 2024-03-22 18:24 54阅读 0赞

### RocketMQ延时消息使用 ###

#### 生产者实例 ####

@Component
    public class DelayProducer {
        
        @Value("${rocketmq.producer.group}")
        private String producerGroup; // 生产者分组
        @Value("${rocketmq.name-server}")
        private String nameServer; // 服务地址
        @Value("${rocketmq.consumer.topic}")
        private String topic; // 生产者主题
        // RocketMQ(延时消息)生产者
        /*
        * tag:
        * OrderAutoCancel 订单自动取消
        * OrderAutoFinish 订单自动完成
        * OrderAutoReceive 订单自动收货
        * SendExpireWarn       发货超期提醒
        * WillPayExpire    将要超期付款提醒
        * PayExpire        超期付款提醒
        * delayTimeLevel:
        * 2  24小时
        * 3  7天
        * 4  15天
        * 5  57天
        * 6  60天
        * */
        //发送延时消息
        public void sendMsg(String tag,Integer parameter,Object object) throws Exception {
        
            // 实例化消息生产者Producer
            DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
            // 设置NameServer的地址
            producer.setNamesrvAddr(nameServer);
            // 启动Producer实例
            producer.start();
            // RockerMQUtil.TOPIC, RockerMQUtil.TAGS,null, JSON.toJSONString(orderJc).getBytes()
            // 参数一是 topic主题，消费者通过匹配topic进行消费，tags对应的是标签,一个topic有多个tags，keys是标识，body对应的是实体
            Message msg = new Message(topic
                    , tag,null, JSON.toJSONString(object).getBytes());
            //"5s 1d 7d 15d 57d 60d"
            //设置消息延迟级别为6，也就是延迟60天。
            Integer delayTimeLevel = null;
            if(parameter == 24){
         // 如果延时时间为24小时，那设置他的等级为2
                delayTimeLevel = 2;
            }else if(parameter == 7){
         // 如果延时时间为7天，那设置他的等级为3
                delayTimeLevel = 3;
            }else if(parameter == 15){
         // 如果延时时间为15天，那设置他的等级为4
                delayTimeLevel = 4;
            }else if(parameter == 57){
         // 如果延时时间为57天，那设置他的等级为5
                delayTimeLevel = 5;
            }else if(parameter == 60){
         // 如果延时时间为60天，那设置他的等级为6
                delayTimeLevel = 6;
            }else if(parameter ==30){
        
                delayTimeLevel = 7;
            }else{
        
                delayTimeLevel = 1; // 如果都不是，那设置延时时间为5秒，等级为1
            }
            msg.setDelayTimeLevel(delayTimeLevel);
            // 发送消息到一个Broker
            producer.send(msg);
            // 如果不再发送消息，关闭Producer实例。
            producer.shutdown();
        }
    
    }

#### 消费者实例 ####

@Component
    @RocketMQMessageListener(topic = RockerMQUtil.TOPIC, consumerGroup = RockerMQUtil.CONSUMERGROUP,
            messageModel = MessageModel.BROADCASTING)// 使用广播模式进行匹配
    public class Consumer implements RocketMQListener<MessageExt> {
        
        @Resource
        private IOrderService orderService;
        @Override
        public void onMessage(MessageExt message) {
        
            SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss.SSS");
            // 实例化消费者
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(RockerMQUtil.CONSUMERGROUP);
            // 设置NameServer的地址
            consumer.setNamesrvAddr(RockerMQUtil.NAMESRVADDR);
            // 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息
            try {
        
                consumer.subscribe(RockerMQUtil.TOPIC, "*");
            } catch (MQClientException e) {
        
                e.printStackTrace();
            }
            // 输出消息被消费的时间
            System.out.printf("%s %s Receive New Messages:%n"
                        , sdf.format(new Date())
                        , Thread.currentThread().getName());
             // 注册回调实现类来处理从broker拉取回来的消息
            String body=new String(message.getBody());
            OrderUtil entity = JSON.parseObject(body,  OrderUtil.class);
            Order order = orderService.getOne(new QueryWrapper<Order>().eq("id",entity.getId())
                    .select("status","id"));
            if(order.getStatus() == 5){
         // 在线付款订单超时未付款，取消订单
                orderService.update(new UpdateWrapper<Order>().eq("id",entity.getId())
                        .set("status",7));
                System.out.println("订单取消成功");
            }
            System.out.printf("\tMsg Id: %s%n", message.getMsgId());
            System.out.printf("\tBody: %s%n", new String(message.getBody()));
            // 启动消费者实例
            try {
        
                consumer.start();
            } catch (MQClientException e) {
        
                e.printStackTrace();
            }
            System.out.println("Consumer Started.");
        }
    }

#### 自定义延时时间 ####

在服务器端（rocketmq-broker端）的属性配置文件broker.conf中加入以下行：

messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h  
描述了各级别与延时时间的对应映射关系。

> 1.  这个配置项配置了从1级开始，各级延时的时间，可以修改这个指定级别的延时时间；
> 2.  时间单位支持：s、m、h、d，分别表示秒、分、时、天；
> 3.  默认值就是上面声明的，可手工调整；
> 4.  默认值已够用，不建议修改这个值。