Shulou(Shulou.com)06/02 Report--

This article mainly introduces "the realization method of RocketMQ delay message". In the daily operation, I believe that many people have doubts about the realization method of RocketMQ delay message. The editor consulted all kinds of materials and sorted out simple and easy-to-use operation methods. I hope it will be helpful to answer the doubts about "implementation method of RocketMQ delay message". Next, please follow the editor to study!

Delayed message means that the message is not immediately visible to the consumer after it is sent, but delivered to the consumer at the time specified by the user. For example, we now send a message with a delay of 30 seconds, which is sent to the server immediately after it is sent, but the server does not deliver the message to the consumer until 30 seconds later.

RocketMQ delays the delivery of messages to consumers through the configured delay level. Different delay levels correspond to different delay times and can be configured. There are 18 default delay levels, which are 1s, 5s, 10s, 30s, 1m, 2m, 3m, 4m, 6m, 7m, 8m, 9m, 10m, 30m, 1h, 2h, s, m, m, minutes, h, h and d days.

The source code MessageStoreConfig.java is defined as follows:

Private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"

You can configure messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h in brocker and customize its time level.

1. Code verification

Premise: first start the consumer to wait for the message to be sent, send the message first, and it will take time for the consumer to start, which will affect the test results.

Producer Producerpublic class DelayProducer {public static void main (String [] args) throws MQClientException, InterruptedException {DefaultMQProducer producer = new DefaultMQProducer ("producer_test"); producer.setNamesrvAddr ("10.10.12.203 String 9876"); producer.start () SimpleDateFormat sd = new SimpleDateFormat ("YYYY-MM-dd HH:mm:ss"); for (int I = 0; I

< 10; i++) { try { //构建消息 Message msg = new Message("TopicTest" /* Topic */, "TagA" /* Tag */, ("延迟消息" + i).getBytes(RemotingHelper.DEFAULT_CHARSET) ); //延时的级别为3 对应的时间为10s 就是发送后延时10S在把消息投递出去 msg.setDelayTimeLevel(3); SendResult sendResult = producer.send(msg); System.out.printf("%s%n", sd.format(new Date())+" == "+sendResult); } catch (Exception e) { e.printStackTrace(); Thread.sleep(1000); } } producer.shutdown(); }} 查看结果：

1.2.Consumer Consumerpublic class DelayConsumer {public static void main (String [] args) {try {DefaultMQPushConsumer consumer = new DefaultMQPushConsumer (); consumer.setConsumerGroup ("consumer_delay"); consumer.setNamesrvAddr ("10.10.12.203 String 9876") Consumer.subscribe ("TopicTest", "*") Consumer.registerMessageListener (new MessageListenerConcurrently () {@ Override public ConsumeConcurrentlyStatus consumeMessage (List paramList) ConsumeConcurrentlyContext paramConsumeConcurrentlyContext) {try {for (MessageExt msg: paramList) {String msgbody = new String (msg.getBody (), "utf-8") SimpleDateFormat sd = new SimpleDateFormat ("YYYY-MM-dd HH:mm:ss"); System.out.println ("receive time:" + sd.format (new Date ()) + "= MessageBody:" + msgbody) / / output message content}} catch (Exception e) {e.printStackTrace (); return ConsumeConcurrentlyStatus.RECONSUME_LATER / / try again later} return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; / / consumption success}}); consumer.start () System.out.println ("DelayConsumer=== started successfully!");} catch (Exception e) {/ / TODO Auto-generated catch block e.printStackTrace ();}

View the results:

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.