5. rocketmq延迟消息队列

今天药忘吃喽~ 2023-07-01 11:56 19阅读 0赞

# 什么是延迟消息队列 #

对于消息中间件来说，producer将消息发送到mq的服务器，但并不期望这条消息马上被消费，而是推迟到当前时间点之后的某个时间点后再投递到queue中让consumer进行消费。

也可以认为是定时消息。

延迟消息的使用场景很多，一种比较常见的场景就是在电商系统中，订单创建后，会有一个等待用户支付的时间窗口，一般为30分钟，30分钟后consumer收到这条订单消息，然后程序去订单表中检查当前这条订单的支付状态，如果是未支付的状态，则自动清理掉。

延迟消息原理是采用`时间轮`，阿里开源的rocketmq的延迟消息是阉割过的，只支持18个级别的延迟时间。采用时间轮算法可以做到任意级别的延迟。关于时间轮算法后续再分析。

# 生产者 #

生产者关键一行代码就是`message.setDelayTimeLevel(3);`设置一下消息的延迟等级。

public static void main(String[] args) throws Exception { 
            //Instantiate with a producer group name.
            DefaultMQProducer producer = new DefaultMQProducer("example_group_name");
            producer.setNamesrvAddr("node1:9876");
            //Launch the instance.
            producer.start();
    
            Message message = new Message("TopicTest", ("this is a delay message:" + "hhhhhhhhhh").getBytes());
    
            //"1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"
            message.setDelayTimeLevel(3);//3指上面第三个策略，是10秒。
            producer.send(message, new SendCallback() { 
    
                //消息发送成功回调
                @Override
                public void onSuccess(SendResult sendResult) { 
                    log.info(sendResult.toString());
                }
    
                //消息异常回调
                @Override
                public void onException(Throwable e) { 
                    e.printStackTrace();
                    //补偿机制，根据业务情况进行使用，看是否进行重试
                }
            });
    
    
            TimeUnit.SECONDS.sleep(3);
    
            producer.shutdown();
        }

# 消费者 #

消费者这里还是借用上一篇文章中的普通消费者。

public static void main(String[] args) throws Exception { 
            normal();//普通消费
    
    // order();//顺序消费
        }
    
        private static void normal() throws MQClientException { 
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("example_group_name");
            consumer.setNamesrvAddr("node1:9876");
    
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
    
            consumer.subscribe("TopicTest", "*");
            consumer.setConsumeThreadMin(3);
            consumer.setConsumeThreadMax(6);
    
            consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> { 
                for (MessageExt msg : msgs) { 
    // System.out.println("收到消息," + new String(msg.getBody()));
                    System.out.println("queueId:"+msg.getQueueId()+",orderId:"+new String(msg.getBody())+",i:"+msg.getKeys());
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
    
            consumer.start();
    
            System.out.printf("Consumer Started.%n");
        }

# 运行 #

先启动消费者，等待接收消息。  
然后启动发送者，发送消息，可以看到过了10秒钟后，消费者可以消费到消息。