rocketmq延时消息

Posted 2020-12-27 peachyy

rocketmq提供一种延时消息的解决方案，就是在特定的时间到了，消息才会被投递出去供consumer消费。

总体来是简单的场景是满足了，但是需要注意的是延时的时间是需要按照默认配置的延时级别去配置的，而不是随意设置消息的延时时间。

如果想不受延时级别的约束 可以参考之前的一遍文章http://blog.seoui.com/2017/08/19/delayqueue/

默认的延迟级别

 

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

这个配置下标从1开始 比如级别2是延时5秒、级别5是延时1分钟。默认配置在不满足需求的情况下，可以在broker配置文件加入messageDelayLevel参数覆盖默认的延时级别配置。

示例

和普通的消息不同之处在于Producer在发送消息的时候 需要设置message.setDelayTimeLevel();延迟级别方法。其他参数和消费端的写法并与不同之处。

Producer

public class ScheduledMessageProducer {
    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = 
        new DefaultMQProducer("ExampleProducerGroup");
        producer.start();
        int totalMessagesToSend = 100;
        for (int i = 0; i < totalMessagesToSend; i++) {
            Message message = new Message("TestTopic",
            ("Hello scheduled message " + i).getBytes());
         //延时的级别为3 对应的时间为10s 就是发送后延时10S在把消息投递出去
            message.setDelayTimeLevel(3);
            producer.send(message);
        }
        producer.shutdown();
    }
}

Consumer

public class ScheduledMessageConsumer {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = 
        new DefaultMQPushConsumer("ExampleConsumer");
        consumer.subscribe("TestTopic", "*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages,
         ConsumeConcurrentlyContext context) {
                for (MessageExt message : messages) {
                    System.out.println("ok!");
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
    }
}