support simple consumer supports subscribing to multiple topics

Author: panzhi33

rocketmq-v5-client-spring-boot-samples/rocketmq-v5-client-consume-simple-subscribe-muliti-topic-demo/src/main/java/org/apache/rocketmq/springboot/ExtRocketMQTemplate.java

@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.springboot;
+
+import org.apache.rocketmq.client.annotation.ExtConsumerResetConfiguration;
+import org.apache.rocketmq.client.core.RocketMQClientTemplate;
+
+@ExtConsumerResetConfiguration(subscriptionExpressions = {
+        @ExtConsumerResetConfiguration.FilterExpression(topic = "demo-topic", tag = "tagA", filterExpressionType = "tag"),
+        @ExtConsumerResetConfiguration.FilterExpression(topic = "demo-topic2", tag = "tagB", filterExpressionType = "tag")
+})
+public class ExtRocketMQTemplate extends RocketMQClientTemplate {
+}

rocketmq-v5-client-spring-boot-samples/rocketmq-v5-client-consume-simple-subscribe-muliti-topic-demo/src/main/java/org/apache/rocketmq/springboot/V5SimpleConsumerConsumerApplication.java

@@ -32,14 +32,20 @@ public class V5SimpleConsumerConsumerApplication implements CommandLineRunner {
     @Resource
     private RocketMQClientTemplate rocketMQClientTemplate;
 
+    @Resource
+    private ExtRocketMQTemplate extRocketMQTemplate;
+
     public static void main(String[] args) {
         SpringApplication.run(V5SimpleConsumerConsumerApplication.class, args);
     }
 
     @Override
     public void run(String... args) throws Exception {
-        for (int i = 0; i < 10; i++) {
-            List<MessageView> messageList = rocketMQClientTemplate.receive(10, Duration.ofSeconds(60));
+        while (true){
+            List<MessageView> messageList = extRocketMQTemplate.receive(10, Duration.ofSeconds(10));
+            System.out.println(messageList);
+
+            messageList = rocketMQClientTemplate.receive(10, Duration.ofSeconds(10));
             System.out.println(messageList);
         }
     }

rocketmq-v5-client-spring-boot-samples/rocketmq-v5-client-consume-simple-subscribe-muliti-topic-demo/src/main/resources/application.properties

@@ -13,12 +13,13 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-rocketmq.simple-consumer.endpoints=localhost:8081
-rocketmq.simple-consumer.consumer-group=localhost:8081
+rocketmq.simple-consumer.endpoints=localhost:8080
+rocketmq.simple-consumer.consumer-group=test-group
 rocketmq.simple-consumer.subscription-expressions.demo-topic.tag=tagA
 rocketmq.simple-consumer.subscription-expressions.demo-topic.filter-expression-type=tag
 rocketmq.simple-consumer.subscription-expressions.demo-topic2.tag=tagB
 rocketmq.simple-consumer.subscription-expressions.demo-topic2.filter-expression-type=tag
 #rocketmq.simple-consumer.access-key=
 #rocketmq.simple-consumer.secret-key=
+#rocketmq.simple-consumer.namespace=
 

rocketmq-v5-client-spring-boot/pom.xml

@@ -93,6 +93,10 @@
             <artifactId>junit</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>com.alibaba</groupId>
+            <artifactId>fastjson</artifactId>
+        </dependency>
     </dependencies>
 
 

rocketmq-v5-client-spring-boot/src/main/java/org/apache/rocketmq/client/annotation/ExtConsumerResetConfiguration.java

@@ -91,4 +91,28 @@
      * The namespace of consumer.
      */
     String namespace() default "";
+
+    /**
+     * subscribing to multiple topics
+     */
+    FilterExpression[] subscriptionExpressions() default {};
+
+    @Retention(RetentionPolicy.RUNTIME)
+    @Target({})
+    @interface FilterExpression {
+        /**
+         * Topic name of consumer.
+         */
+        String topic();
+
+        /**
+         * Tag of consumer.
+         */
+        String tag() default "*";
+
+        /**
+         * The type of filter expression
+         */
+        String filterExpressionType() default "tag";
+    }
 }

rocketmq-v5-client-spring-boot/src/main/java/org/apache/rocketmq/client/autoconfigure/ExtConsumerResetConfiguration.java

@@ -16,6 +16,7 @@
  */
 package org.apache.rocketmq.client.autoconfigure;
 
+import com.alibaba.fastjson.JSON;
 import org.apache.rocketmq.client.support.RocketMQMessageConverter;
 import org.apache.rocketmq.client.support.RocketMQUtil;
 import org.apache.rocketmq.client.apis.ClientConfiguration;
@@ -40,7 +41,7 @@
 import org.springframework.util.StringUtils;
 
 import java.time.Duration;
-import java.util.Collections;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
 import java.util.stream.Collectors;
@@ -110,30 +111,44 @@ private SimpleConsumerInfo createConsumer(
         SimpleConsumerBuilder simpleConsumerBuilder) {
         RocketMQProperties.SimpleConsumer simpleConsumer = rocketMQProperties.getSimpleConsumer();
         String consumerGroupName = resolvePlaceholders(annotation.consumerGroup(), simpleConsumer.getConsumerGroup());
-        String topicName = resolvePlaceholders(annotation.topic(), simpleConsumer.getTopic());
         String accessKey = resolvePlaceholders(annotation.accessKey(), simpleConsumer.getAccessKey());
         String secretKey = resolvePlaceholders(annotation.secretKey(), simpleConsumer.getSecretKey());
         String endPoints = resolvePlaceholders(annotation.endpoints(), simpleConsumer.getEndpoints());
         String namespace = resolvePlaceholders(annotation.namespace(), simpleConsumer.getNamespace());
-        String tag = resolvePlaceholders(annotation.tag(), simpleConsumer.getTag());
-        String filterExpressionType = resolvePlaceholders(annotation.filterExpressionType(), simpleConsumer.getFilterExpressionType());
         Duration requestTimeout = Duration.ofSeconds(annotation.requestTimeout());
         int awaitDuration = annotation.awaitDuration();
         Boolean sslEnabled = simpleConsumer.isSslEnabled();
-        Assert.hasText(topicName, "[topic] must not be null");
         ClientConfiguration clientConfiguration = RocketMQUtil.createClientConfiguration(accessKey, secretKey, endPoints, requestTimeout, sslEnabled, namespace);
-        FilterExpression filterExpression = RocketMQUtil.createFilterExpression(tag, filterExpressionType);
         Duration duration = Duration.ofSeconds(awaitDuration);
         simpleConsumerBuilder.setClientConfiguration(clientConfiguration);
         if (StringUtils.hasLength(consumerGroupName)) {
             simpleConsumerBuilder.setConsumerGroup(consumerGroupName);
         }
         simpleConsumerBuilder.setAwaitDuration(duration);
-        if (Objects.nonNull(filterExpression)) {
-            simpleConsumerBuilder.setSubscriptionExpressions(Collections.singletonMap(topicName, filterExpression));
+
+        Map<String, FilterExpression> subscriptionExpressions = new HashMap<>();
+        org.apache.rocketmq.client.annotation.ExtConsumerResetConfiguration.FilterExpression[] filterExpressions = annotation.subscriptionExpressions();
+        if (filterExpressions.length > 0) {
+            for (org.apache.rocketmq.client.annotation.ExtConsumerResetConfiguration.FilterExpression expression : filterExpressions) {
+                Assert.hasText(expression.topic(), "[topic] must not be null");
+                FilterExpression filterExpression = RocketMQUtil.createFilterExpression(expression.tag(), expression.filterExpressionType());
+                if (Objects.nonNull(filterExpression)) {
+                    subscriptionExpressions.put(expression.topic(), filterExpression);
+                }
+            }
+        } else {
+            String topicName = resolvePlaceholders(annotation.topic(), simpleConsumer.getTopic());
+            Assert.hasText(topicName, "[topic] must not be null");
+            String tag = resolvePlaceholders(annotation.tag(), simpleConsumer.getTag());
+            String filterExpressionType = resolvePlaceholders(annotation.filterExpressionType(), simpleConsumer.getFilterExpressionType());
+            FilterExpression filterExpression = RocketMQUtil.createFilterExpression(tag, filterExpressionType);
+            if (Objects.nonNull(filterExpression)) {
+                subscriptionExpressions.put(topicName, filterExpression);
+            }
         }
+        simpleConsumerBuilder.setSubscriptionExpressions(subscriptionExpressions);
 
-        return new SimpleConsumerInfo(consumerGroupName, topicName, endPoints, namespace, tag, filterExpressionType, requestTimeout, awaitDuration, sslEnabled);
+        return new SimpleConsumerInfo(consumerGroupName, endPoints, namespace, requestTimeout, awaitDuration, sslEnabled, subscriptionExpressions);
     }
 
     private String resolvePlaceholders(String text, String defaultValue) {
@@ -144,47 +159,40 @@ private String resolvePlaceholders(String text, String defaultValue) {
     static class SimpleConsumerInfo {
         String consumerGroup;
 
-        String topicName;
-
         String endPoints;
 
         String namespace;
 
-        String tag;
-
-        String filterExpressionType;
-
         Duration requestTimeout;
 
         int awaitDuration;
 
         Boolean sslEnabled;
 
-        public SimpleConsumerInfo(String consumerGroupName, String topicName, String endPoints, String namespace,
-            String tag, String filterExpressionType, Duration requestTimeout, int awaitDuration, Boolean sslEnabled) {
+        Map<String, FilterExpression> subscriptionExpressions;
+
+        public SimpleConsumerInfo(String consumerGroupName, String endPoints, String namespace, Duration requestTimeout,
+                                  int awaitDuration, Boolean sslEnabled, Map<String, FilterExpression> subscriptionExpressions) {
             this.consumerGroup = consumerGroupName;
-            this.topicName = topicName;
             this.endPoints = endPoints;
             this.namespace = namespace;
-            this.tag = tag;
-            this.filterExpressionType = filterExpressionType;
             this.requestTimeout = requestTimeout;
             this.awaitDuration = awaitDuration;
             this.sslEnabled = sslEnabled;
+            this.subscriptionExpressions = subscriptionExpressions;
         }
 
-        @Override public String toString() {
+        @Override
+        public String toString() {
             return "SimpleConsumerInfo{" +
-                "consumerGroup='" + consumerGroup + '\'' +
-                ", topicName='" + topicName + '\'' +
-                ", endPoints='" + endPoints + '\'' +
-                ", namespace='" + namespace + '\'' +
-                ", tag='" + tag + '\'' +
-                ", filterExpressionType='" + filterExpressionType + '\'' +
-                ", requestTimeout(seconds)=" + requestTimeout.getSeconds() +
-                ", awaitDuration=" + awaitDuration +
-                ", sslEnabled=" + sslEnabled +
-                '}';
+                    "consumerGroup='" + consumerGroup + '\'' +
+                    ", endPoints='" + endPoints + '\'' +
+                    ", namespace='" + namespace + '\'' +
+                    ", requestTimeout=" + requestTimeout +
+                    ", awaitDuration=" + awaitDuration +
+                    ", sslEnabled=" + sslEnabled +
+                    ", subscriptionExpressions=" + JSON.toJSONString(subscriptionExpressions) +
+                    '}';
         }
     }
 }

rocketmq-v5-client-spring-boot/src/main/java/org/apache/rocketmq/client/autoconfigure/RocketMQAutoConfiguration.java

@@ -42,6 +42,7 @@
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
 import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
 import org.springframework.util.StringUtils;
 
 import java.time.Duration;
@@ -118,7 +119,7 @@ public SimpleConsumerBuilder simpleConsumerBuilder(RocketMQProperties rocketMQPr
         }
 
         // Set the subscription for the consumer.
-        if (simpleConsumer.getSubscriptionExpressions().isEmpty()) {
+        if (CollectionUtils.isEmpty(simpleConsumer.getSubscriptionExpressions())) {
             FilterExpression filterExpression = RocketMQUtil.createFilterExpression(simpleConsumer.getTag(), simpleConsumer.getFilterExpressionType());
             if (Objects.nonNull(filterExpression)) {
                 simpleConsumerBuilder.setSubscriptionExpressions(Collections.singletonMap(simpleConsumer.getTopic(), filterExpression));