support alibaba cloud rocketmq

Author: flystar32

mse-simple-demo/A/pom.xml

@@ -59,6 +59,11 @@
             <artifactId>rocketmq-client</artifactId>
             <version>${rocketmq.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-acl</artifactId>
+            <version>${rocketmq.version}</version>
+        </dependency>
 
         <!--  Dubbo -->
         <dependency>

mse-simple-demo/A/src/main/java/com/alibabacloud/mse/demo/a/mq/RocketMqConfiguration.java

@@ -2,11 +2,15 @@
 
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.acl.common.AclClientRPCHook;
+import org.apache.rocketmq.acl.common.SessionCredentials;
 import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
 import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
+import org.apache.rocketmq.remoting.RPCHook;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.cloud.commons.util.InetUtils;
 import org.springframework.context.annotation.Bean;
@@ -27,6 +31,11 @@ public class RocketMqConfiguration {
     @Value("${rocketmq.consumer.topic}")
     private String topic;
 
+    @Value("${middleware.mq.ak:}")
+    private String ak;
+    @Value("${middleware.mq.sk:}")
+    private String sk;
+
     @Autowired
     private RestTemplate restTemplate;
 
@@ -43,16 +52,21 @@ public class RocketMqConfiguration {
     @Bean(initMethod = "start", destroyMethod = "shutdown")
     public DefaultMQPushConsumer mqPushConsumer() throws MQClientException {
         log.info("正在启动rocketMq的consumer");
-        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
+
+        DefaultMQPushConsumer consumer;
+        if (StringUtils.isNotEmpty(this.ak) && StringUtils.isNotEmpty(this.sk)) {
+            log.info("middleware.mq.ak 不为空,MQ ACL信息已填充");
+            RPCHook rpcHook = new AclClientRPCHook(new SessionCredentials(this.ak, this.sk));
+            consumer = new DefaultMQPushConsumer(this.groupName, rpcHook, new AllocateMessageQueueAveragely());
+        } else {
+            log.info("middleware.mq.ak 为空");
+            consumer = new DefaultMQPushConsumer(groupName);
+        }
         consumer.setNamesrvAddr(nameSrvAddr);
         consumer.subscribe(topic, "*");
         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
 
-        MqConsumer mqConsumer = new MqConsumer(
-                restTemplate,
-                inetUtils,
-                serviceTag
-        );
+        MqConsumer mqConsumer = new MqConsumer(restTemplate, inetUtils, serviceTag);
         consumer.registerMessageListener(mqConsumer);
         log.info("完成启动rocketMq的consumer,subscribe:{}", topic);
         return consumer;

mse-simple-demo/C/pom.xml

@@ -49,6 +49,11 @@
             <artifactId>rocketmq-client</artifactId>
             <version>${rocketmq.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-acl</artifactId>
+            <version>${rocketmq.version}</version>
+        </dependency>
 
 
         <!--  Dubbo -->

mse-simple-demo/C/src/main/java/com/alibabacloud/mse/demo/c/mq/RocketMqConfiguration.java

@@ -3,7 +3,11 @@
 
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.acl.common.AclClientRPCHook;
+import org.apache.rocketmq.acl.common.SessionCredentials;
 import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.remoting.RPCHook;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -19,6 +23,11 @@ public class RocketMqConfiguration {
     @Value("${rocketmq.consumer.group}")
     private String groupName;
 
+    @Value("${middleware.mq.ak:}")
+    private String ak;
+    @Value("${middleware.mq.sk:}")
+    private String sk;
+
     static {
         System.setProperty("rocketmq.client.log.loadconfig", "false");
     }
@@ -27,7 +36,16 @@ public class RocketMqConfiguration {
     public DefaultMQProducer getBaseProducer() throws Exception {
         log.info("正在启动rocketMq的producer");
 
-        DefaultMQProducer baseProducer = new DefaultMQProducer();
+        DefaultMQProducer baseProducer;
+        if (StringUtils.isNotEmpty(this.ak) && StringUtils.isNotEmpty(this.sk)) {
+            log.info("middleware.mq.ak 不为空,MQ ACL信息已填充");
+            RPCHook rpcHook = new AclClientRPCHook(new SessionCredentials(this.ak, this.sk));
+            baseProducer = new DefaultMQProducer(rpcHook);
+        } else {
+            log.info("middleware.mq.ak 为空");
+            baseProducer = new DefaultMQProducer();
+        }
+
         baseProducer.setProducerGroup(groupName);
         baseProducer.setNamesrvAddr(nameSrvAddr);
         baseProducer.start();