合并版本

sermant-io · Jan 3, 2025 · 97db7db · 97db7db
1 parent 1734f10
commit 97db7db
Show file tree

Hide file tree

Showing 6 changed files with 221 additions and 528 deletions.
diff --git a/.github/workflows/message_gray_integration_test.yml b/.github/workflows/message_gray_integration_test.yml
diff --git a/...src/main/java/io/sermant/demo/grayscale/rocketmq/consumer/RocketMqConsumerController.java b/...src/main/java/io/sermant/demo/grayscale/rocketmq/consumer/RocketMqConsumerController.java
@@ -59,15 +59,15 @@ public String clearMessageCount() {
     }
 
     /**
-     * init push consumer
+     * init consumer
      *
      * @param consumerType consumerType
      * @return init status
      */
     @GetMapping("/initConsumer")
     public String initConsumer(@RequestParam("consumerType") String consumerType) {
         if ("PUSH".equals(consumerType)) {
-            pushConsumer.initPushConsume(topic + "-PUSH", nameServer);
+            pushConsumer.initPushConsumer(topic + "-PUSH", nameServer);
         } else if ("PULL".equals(consumerType)) {
             pullConsumer.initPullConsumer(nameServer, topic + "-PULL");
         } else {
@@ -76,6 +76,24 @@ public String initConsumer(@RequestParam("consumerType") String consumerType) {
         return "success";
     }
 
+    /**
+     * shutdown consumer
+     *
+     * @param consumerType consumerType
+     * @return status
+     */
+    @GetMapping("/shutdownConsumer")
+    public String shutdownConsumer(@RequestParam("consumerType") String consumerType) {
+        if ("PUSH".equals(consumerType)) {
+            pushConsumer.shutdownPushConsumer();
+        } else if ("PULL".equals(consumerType)) {
+            pullConsumer.shutdownPullConsumer();
+        } else {
+            litePullConsumer.shutdownLitePullConsumer();
+        }
+        return "success";
+    }
+
     /**
      * pull message
      *

diff --git a/...o/src/main/java/io/sermant/demo/grayscale/rocketmq/consumer/RocketMqLitePullConsumer.java b/...o/src/main/java/io/sermant/demo/grayscale/rocketmq/consumer/RocketMqLitePullConsumer.java
@@ -82,6 +82,16 @@ public Map<String, Object> getMessageCount(String mqTopic, String mqAddress) {
         return RocketMqMessageUtils.getMessageCount();
     }
 
+    /**
+     * shutdown consumer
+     */
+    public void shutdownLitePullConsumer() {
+        executorService.shutdown();
+        if (litePullConsumer != null) {
+            litePullConsumer.shutdown();
+        }
+    }
+
     /**
      * lite pull consumer runnable
      *

diff --git a/...-demo/src/main/java/io/sermant/demo/grayscale/rocketmq/consumer/RocketMqPullConsumer.java b/...-demo/src/main/java/io/sermant/demo/grayscale/rocketmq/consumer/RocketMqPullConsumer.java
@@ -99,6 +99,16 @@ private static long getMessageQueueOffSet(MessageQueue mq) {
         return OFF_SET_TABLE.getOrDefault(mq, 0L);
     }
 
+    /**
+     * shutdown consumer
+     */
+    public void shutdownPullConsumer() {
+        executorService.shutdown();
+        if (pullConsumer != null) {
+            pullConsumer.shutdown();
+        }
+    }
+
     /**
      * pull consumer runnable
      *

diff --git a/...-demo/src/main/java/io/sermant/demo/grayscale/rocketmq/consumer/RocketMqPushConsumer.java b/...-demo/src/main/java/io/sermant/demo/grayscale/rocketmq/consumer/RocketMqPushConsumer.java
@@ -47,7 +47,7 @@ public class RocketMqPushConsumer {
      * @param mqTopic topic
      * @param mqAddress address
      */
-    public void initPushConsume(String mqTopic, String mqAddress) {
+    public void initPushConsumer(String mqTopic, String mqAddress) {
         try {
             if (pushConsumer == null) {
                 pushConsumer = new DefaultMQPushConsumer("default");
@@ -76,8 +76,17 @@ public void initPushConsume(String mqTopic, String mqAddress) {
      */
     public Map<String, Object> getMessageCount(String mqTopic, String mqAddress) {
         if (pushConsumer == null) {
-            initPushConsume(mqTopic, mqAddress);
+            initPushConsumer(mqTopic, mqAddress);
         }
         return RocketMqMessageUtils.getMessageCount();
     }
+
+    /**
+     * shutdown consumer
+     */
+    public void shutdownPushConsumer() {
+        if (pushConsumer != null) {
+            pushConsumer.shutdown();
+        }
+    }
 }