添加RocketMQ 定时任务

coderbruis · coderbruis · commit 15ef0f45f99e · 2021-07-24T11:56:55.000+08:00
diff --git a/rocketmqdemo/README.md b/rocketmqdemo/README.md
@@ -4,4 +4,6 @@
 
 - demo02 普通生产者消费者（异步发送消息）
 
-- demo03 顺序消费
+- demo03 顺序消费
+
+- demo04 延时消息
diff --git a/rocketmqdemo/src/main/java/com/bruis/rocketmqdemo/demo03/Producer.java b/rocketmqdemo/src/main/java/com/bruis/rocketmqdemo/demo03/Producer.java
@@ -1,8 +1,10 @@
 package com.bruis.rocketmqdemo.demo03;
 
 import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.client.producer.MessageQueueSelector;
 import org.apache.rocketmq.client.producer.SendResult;
 import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageQueue;
 
 import java.text.SimpleDateFormat;
 import java.util.ArrayList;
@@ -42,11 +44,20 @@ public static void main(String[] args) throws Exception {
             String body = dateStr + " Hello RocketMQ " + orderList.get(i);
             Message msg = new Message(TOPIC_NAME, tags[i % tags.length], "KEY" + i, body.getBytes());
 
-            SendResult sendResult = producer.send(msg, (mqs, msg1, arg) -> {
-                Long id = (Long) arg;  //根据订单id选择发送queue
-                long index = id % mqs.size();
-                return mqs.get((int) index);
-            }, orderList.get(i).getOrderId());//订单id
+//            SendResult sendResult = producer.send(msg, (mqs, msg1, arg) -> {
+//                Long id = (Long) arg;  //根据订单id选择发送queue
+//                long index = id % mqs.size();
+//                return mqs.get((int) index);
+//            }, orderList.get(i).getOrderId());//订单id
+
+            SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
+                @Override
+                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
+                    Long id = (Long) arg;
+                    long index = id % mqs.size();
+                    return mqs.get((int) index);
+                }
+            }, orderList.get(i).getOrderId());
 
            System.out.println(String.format("SendResult status:%s, queueId:%d, body:%s",
                sendResult.getSendStatus(),
diff --git a/rocketmqdemo/src/main/java/com/bruis/rocketmqdemo/demo04/ScheduledMessageConsumer.java b/rocketmqdemo/src/main/java/com/bruis/rocketmqdemo/demo04/ScheduledMessageConsumer.java
@@ -0,0 +1,39 @@
+package com.bruis.rocketmqdemo.demo04;
+
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.listener.*;
+import org.apache.rocketmq.common.message.MessageExt;
+import java.util.List;
+
+/**
+ *
+ * 延时消息消费者
+ *
+ * @author lhy
+ * @date 2021/7/24
+ */
+public class ScheduledMessageConsumer {
+
+    public static void main(String[] args) throws Exception {
+
+        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(ScheduledMessageProducer.GROUP_NAME);
+        // 订阅指定的topic
+        consumer.subscribe(ScheduledMessageProducer.TOPIC_NAME, "*");
+        consumer.setNamesrvAddr(ScheduledMessageProducer.NAMESRV_ADDRESS);
+        // 注册消息监听者
+        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.getBornTimestamp()) + "ms later");
+                }
+                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+            }
+        });
+        consumer.start();
+    }
+
+}
diff --git a/rocketmqdemo/src/main/java/com/bruis/rocketmqdemo/demo04/ScheduledMessageProducer.java b/rocketmqdemo/src/main/java/com/bruis/rocketmqdemo/demo04/ScheduledMessageProducer.java
@@ -0,0 +1,40 @@
+package com.bruis.rocketmqdemo.demo04;
+
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.common.message.Message;
+
+/**
+ *
+ * 延时消息生产
+ *
+ * @author lhy
+ * @date 2021/7/24
+ */
+public class ScheduledMessageProducer {
+
+    public static final String GROUP_NAME = "scheduled_group";
+
+    public static final String TOPIC_NAME = "scheduled_test_topic";
+
+    public static final String NAMESRV_ADDRESS = "127.0.0.1:9876";
+
+    public static void main(String[] args) throws Exception {
+
+        DefaultMQProducer producer = new DefaultMQProducer(GROUP_NAME);
+        producer.setNamesrvAddr(NAMESRV_ADDRESS);
+        producer.start();
+
+        int totalMessagesToSend = 100;
+
+        for (int i = 0; i < totalMessagesToSend; i++) {
+            Message message = new Message(TOPIC_NAME, ("Hello Scheduled Message " + i).getBytes());
+            // 设置延时等级3,这个消息将在10s之后发送(现在只支持固定的几个时间,详看delayTimeLevel)
+            message.setDelayTimeLevel(3);
+            // 发送消息
+            producer.send(message);
+        }
+
+        producer.shutdown();
+    }
+
+}
