title: 延时消息示例 date: 2017/12/21

categories: 文档翻译

Schedule example

What is scheduled message?

Scheduled messages differ from normal messages in that they won’t be delivered until a provided time later.

延时消息示例

什么是延时消息？

延时消息提供了一种不同于普通消息的实现形式——它们只会在设定的时限到了之后才被递送出去。

Application

  1. Start consumer to wait for incoming subscribed messages

    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 {
    10.   public static void main(String[] args) throws Exception {
    11.       // Instantiate message consumer
    12.       DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ExampleConsumer");
    13.       // Subscribe topics
    14.       consumer.subscribe("TestTopic", "*");
    15.       // Register message listener
    16.       consumer.registerMessageListener(new MessageListenerConcurrently() {
    17.           @Override
    18.           public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context) {
    19.               for (MessageExt message : messages) {
    20.                   // Print approximate delay time period
    21.                   System.out.println("Receive message[msgId=" + message.getMsgId() + "] "
    22.                           + (System.currentTimeMillis() - message.getStoreTimestamp()) + "ms later");
    23.               }
    24.               return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    25.           }
    26.       });
    27.       // Launch consumer
    28.       consumer.start();
    29.   }
    30. }

应用

  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 {
    10.   public static void main(String[] args) throws Exception {
    11.       // Instantiate message consumer
    12.       DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ExampleConsumer");
    13.       // Subscribe topics
    14.       consumer.subscribe("TestTopic", "*");
    15.       // Register message listener
    16.       consumer.registerMessageListener(new MessageListenerConcurrently() {
    17.           @Override
    18.           public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context) {
    19.               for (MessageExt message : messages) {
    20.                   // Print approximate delay time period
    21.                   System.out.println("Receive message[msgId=" + message.getMsgId() + "] "
    22.                           + (System.currentTimeMillis() - message.getStoreTimestamp()) + "ms later");
    23.               }
    24.               return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    25.           }
    26.       });
    27.       // Launch consumer
    28.       consumer.start();
    29.   }
    30. }

  2. Send scheduled messages

    1. import org.apache.rocketmq.client.producer.DefaultMQProducer;
    2. import org.apache.rocketmq.common.message.Message;
    4. public class ScheduledMessageProducer {
    6.   public static void main(String[] args) throws Exception {
    7.       // Instantiate a producer to send scheduled messages
    8.       DefaultMQProducer producer = new DefaultMQProducer("ExampleProducerGroup");
    9.       // Launch producer
    10.       producer.start();
    11.       int totalMessagesToSend = 100;
    12.       for (int i = 0; i < totalMessagesToSend; i++) {
    13.           Message message = new Message("TestTopic", ("Hello scheduled message " + i).getBytes());
    14.           // This message will be delivered to consumer 10 seconds later.
    15.           message.setDelayTimeLevel(3);
    16.           // Send the message
    17.           producer.send(message);
    18.       }
    20.       // Shutdown producer after use.
    21.       producer.shutdown();
    22.   }
    24. }

  3. 发送延时消息

    1. import org.apache.rocketmq.client.producer.DefaultMQProducer;
    2. import org.apache.rocketmq.common.message.Message;
    4. public class ScheduledMessageProducer {
    6.   public static void main(String[] args) throws Exception {
    7.       // Instantiate a producer to send scheduled messages
    8.       DefaultMQProducer producer = new DefaultMQProducer("ExampleProducerGroup");
    9.       // Launch producer
    10.       producer.start();
    11.       int totalMessagesToSend = 100;
    12.       for (int i = 0; i < totalMessagesToSend; i++) {
    13.           Message message = new Message("TestTopic", ("Hello scheduled message " + i).getBytes());
    14.           // This message will be delivered to consumer 10 seconds later.
    15.           message.setDelayTimeLevel(3);
    16.           // Send the message
    17.           producer.send(message);
    18.       }
    20.       // Shutdown producer after use.
    21.       producer.shutdown();
    22.   }
    24. }

  4. Verification You should see messages are consumed about 10 seconds later than their storing time.

  5. 验证 你应该会在消息被存储之后10秒钟看到它们被消费。