3 延时消息样例

3.1 启动消费者等待传入订阅消息

  1. import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
  2. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
  3. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
  4. import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
  5. import org.apache.rocketmq.common.message.MessageExt;
  6. import java.util.List;
  8. public class ScheduledMessageConsumer {
  9.    public static void main(String[] args) throws Exception {
  10.       // 实例化消费者
  11.       DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ExampleConsumer");
  12.       // 订阅Topics
  13.       consumer.subscribe("TestTopic", "*");
  14.       // 注册消息监听者
  15.       consumer.registerMessageListener(new MessageListenerConcurrently() {
  16.           @Override
  17.           public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context) {
  18.               for (MessageExt message : messages) {
  19.                   // Print approximate delay time period
  20.                   System.out.println("Receive message[msgId=" + message.getMsgId() + "] " + (System.currentTimeMillis() - message.getStoreTimestamp()) + "ms later");
  21.               }
  22.               return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  23.           }
  24.       });
  25.       // 启动消费者
  26.       consumer.start();
  27.   }
  28. }

3.2 发送延时消息

  1. import org.apache.rocketmq.client.producer.DefaultMQProducer;
  2. import org.apache.rocketmq.common.message.Message;
  4. public class ScheduledMessageProducer {
  5.    public static void main(String[] args) throws Exception {
  6.       // 实例化一个生产者来产生延时消息
  7.       DefaultMQProducer producer = new DefaultMQProducer("ExampleProducerGroup");
  8.       // 启动生产者
  9.       producer.start();
  10.       int totalMessagesToSend = 100;
  11.       for (int i = 0; i < totalMessagesToSend; i++) {
  12.           Message message = new Message("TestTopic", ("Hello scheduled message " + i).getBytes());
  13.           // 设置延时等级3,这个消息将在10s之后发送(现在只支持固定的几个时间,详看delayTimeLevel)
  14.           message.setDelayTimeLevel(3);
  15.           // 发送消息
  16.           producer.send(message);
  17.       }
  18.        // 关闭生产者
  19.       producer.shutdown();
  20.   }
  21. }

3.3 验证

您将会看到消息的消费比存储时间晚10秒。

3.4 延时消息的使用场景

比如电商里，提交了一个订单就可以发送一个延时消息，1h后去检查这个订单的状态，如果还是未付款就取消订单释放库存。

3.5 延时消息的使用限制

  1. // org/apache/rocketmq/store/config/MessageStoreConfig.java
  3. private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";

现在RocketMq并不支持任意时间的延时，需要设置几个固定的延时等级，从1s到2h分别对应着等级1到18 消息消费失败会进入延时消息队列，消息发送时间与设置的延时等级和重试次数有关，详见代码SendMessageProcessor.java