rocket mq 时间表示例——setDelayTimeLevel

亦凉 2022-06-08 04:52 171阅读 0赞

# 时间表示例 #

# 计划的消息与正常消息不同，因为它们将在以后提供的时间内不会被传送。 #

### 什么是预定的消息？ ###

### 应用 ###

1.  开始消费者等待传入的订阅消息
    
        import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
         import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
         import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
         import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
         import org.apache.rocketmq.common.message.MessageExt;
         import java.util.List;
            
         public class ScheduledMessageConsumer { 
            
             public static void main(String[] args) throws Exception { 
                 // Instantiate message consumer
                 DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ExampleConsumer");
                 // Subscribe topics
                 consumer.subscribe("TestTopic", "*");
                 // Register message listener
                 consumer.registerMessageListener(new MessageListenerConcurrently() { 
                     @Override
                     public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context) { 
                         for (MessageExt message : messages) { 
                             // Print approximate delay time period
                             System.out.println("Receive message[msgId=" + message.getMsgId() + "] "
                                     + (System.currentTimeMillis() - message.getStoreTimestamp()) + "ms later");
                         }
                         return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                     }
                 });
                 // Launch consumer
                 consumer.start();
             }
         }
2.  发送预定的消息
    
        import org.apache.rocketmq.client.producer.DefaultMQProducer;
         import org.apache.rocketmq.common.message.Message;
            
         public class ScheduledMessageProducer { 
            
             public static void main(String[] args) throws Exception { 
                 // Instantiate a producer to send scheduled messages
                 DefaultMQProducer producer = new DefaultMQProducer("ExampleProducerGroup");
                 // Launch producer
                 producer.start();
                 int totalMessagesToSend = 100;
                 for (int i = 0; i < totalMessagesToSend; i++) { 
                     Message message = new Message("TestTopic", ("Hello scheduled message " + i).getBytes());
                     // This message will be delivered to consumer 10 seconds later.
                     message.setDelayTimeLevel(3);
                     // Send the message
                     producer.send(message);
                 }
            
                 // Shutdown producer after use.
                 producer.shutdown();
             }
                
         }
3.  验证
    
    您应该看到消息比他们的存储时间晚了大约10秒钟。