RocketMQ延时消息

RocketMQ延时消息

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参数覆盖默认的延时级别配置。

RocketMQ发送延时消息

和普通的消息不同之处在于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();
    }
}