RocketMQ使用延迟消息

Posted 2022-01-07 乐观男孩

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了RocketMQ使用延迟消息相关的知识，希望对你有一定的参考价值。

说明

1、延时消息不是延迟发送，消息是实时发送的，只是消费者延迟消费
2、延迟消息主要通过对Message设置延迟级别实现，生产者和消费者按照正常逻辑进行生产和消费。

生产端

@Test
    public void sendMessage() throws Exception 
        DefaultMQProducer defaultMQProducer = RocketMqUtil.getDefaultMQProducer();
        Message message = new Message(RocketMqUtil.TOPIC, "schedule",
                "schedule-message".getBytes(Charset.forName("UTF-8")));
        //设置延迟级别，延迟级别≠延迟时间
        message.setDelayTimeLevel(5);
        SendResult sendResult = defaultMQProducer.send(message);
        log.info("发送消息结果：", sendResult.getSendStatus().name());

消费端

@Test
    public void consumer() throws Exception 
        DefaultMQPushConsumer defaultMQPushConsumer = RocketMqUtil.getDefaultMQPushConsumer();
        defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        defaultMQPushConsumer.subscribe(RocketMqUtil.TOPIC, "schedule");
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() 
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) 
                for (MessageExt messageExt : msgs) 
                    log.info("出生时间：，存储时间：，当前时间：，消息内容：",
                            sdf.format(new Date(messageExt.getBornTimestamp())),
                            sdf.format(new Date(messageExt.getStoreTimestamp())),
                            sdf.format(new Date()),
                            new String(messageExt.getBody(), Charset.forName("UTF-8")));
                
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            
        );
        defaultMQPushConsumer.start();
        Thread.sleep(30000L);
        defaultMQPushConsumer.shutdown();

运行结果示例

这里解释一下：
BornTimestamp：消息的出生时间，这是在客户端发送消息前设置（DefaultMQProducerImpl.sendKernelImpl）。
StoreTimestamp：消息的存储时间（并不是指服务端接收的时间），这是在服务端进行设置的（CommitLog.asyncPutMessage）。