update rocketmq configs

examcloud-core-basic-service/src/main/java/cn/com/qmth/examcloud/core/basic/service/report/RocketMqConsumerListener.java

@@ -1,25 +1,17 @@
 package cn.com.qmth.examcloud.core.basic.service.report;
 
-import java.util.Properties;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.alibaba.fastjson.JSON;
-import com.aliyun.openservices.ons.api.Action;
-import com.aliyun.openservices.ons.api.ConsumeContext;
-import com.aliyun.openservices.ons.api.Consumer;
-import com.aliyun.openservices.ons.api.Message;
-import com.aliyun.openservices.ons.api.MessageListener;
-import com.aliyun.openservices.ons.api.ONSFactory;
-import com.aliyun.openservices.ons.api.PropertyKeyConst;
-
 import cn.com.qmth.examcloud.core.basic.service.AdminOperateService;
 import cn.com.qmth.examcloud.reports.commons.bean.AdminOperateReport;
 import cn.com.qmth.examcloud.reports.commons.enums.Tag;
 import cn.com.qmth.examcloud.reports.commons.util.ReportsUtil;
 import cn.com.qmth.examcloud.web.bootstrap.PropertyHolder;
 import cn.com.qmth.examcloud.web.support.SpringContextHolder;
+import com.alibaba.fastjson.JSON;
+import com.aliyun.openservices.ons.api.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Properties;
 
 public class RocketMqConsumerListener {
 
@@ -30,17 +22,19 @@ public class RocketMqConsumerListener {
     private static Properties properties = new Properties();
 
     static {
-        properties.put(PropertyKeyConst.AccessKey, PropertyHolder.getString("$rocketmq-accesskey"));
+        properties.put(PropertyKeyConst.AccessKey, PropertyHolder.getString("examcloud.rocketmq.accesskey"));
         // AccessKeySecret 阿里云身份验证，在阿里云服务器管理控制台创建。
-        properties.put(PropertyKeyConst.SecretKey, PropertyHolder.getString("$rocketmq-secretkey"));
+        properties.put(PropertyKeyConst.SecretKey, PropertyHolder.getString("examcloud.rocketmq.secretkey"));
         // 设置发送超时时间，单位毫秒。
         properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, "3000");
         // 设置 TCP 接入域名，进入控制台的实例详情页面的 TCP 协议客户端接入点区域查看。
-        properties.put(PropertyKeyConst.NAMESRV_ADDR, PropertyHolder.getString("$rocketmq-namesrv-addr"));
+        properties.put(PropertyKeyConst.NAMESRV_ADDR, PropertyHolder.getString("examcloud.rocketmq.namesrv_addr"));
         // 顺序消息消费失败进行重试前的等待时间，单位（毫秒），取值范围: 10 毫秒 ~ 30,000 毫秒
         properties.put(PropertyKeyConst.SuspendTimeMillis, "100");
         // 消息消费失败时的最大重试次数
         properties.put(PropertyKeyConst.MaxReconsumeTimes, "10");
+        // 消费组名称
+        properties.put(PropertyKeyConst.GROUP_ID, PropertyHolder.getString("examcloud.rocketmq.consumer_group"));
     }
 
     public static void start() {
@@ -48,7 +42,6 @@ public class RocketMqConsumerListener {
     }
 
     private static void adminOperate() {
-        properties.put(PropertyKeyConst.GROUP_ID, Tag.ADMIN_OPERATE_INFO.getGroup());
         Consumer consumer = ONSFactory.createConsumer(properties);
         consumer.subscribe(ReportsUtil.getReportTopic(), Tag.ADMIN_OPERATE_INFO.getCode(), new MessageListener() {
 
@@ -63,7 +56,6 @@ public class RocketMqConsumerListener {
                     return Action.ReconsumeLater;
                 }
             }
-
         });
 
         consumer.start();
@@ -73,4 +65,5 @@ public class RocketMqConsumerListener {
         AdminOperateReport r = JSON.parseObject(message, AdminOperateReport.class);
         adminOperateService.saveOperate(r, msgId);
     }
+
 }

examcloud-core-basic-starter/src/main/java/cn/com/qmth/examcloud/core/basic/starter/config/MqConsumerListenerStartup.java

@@ -1,5 +1,7 @@
 package cn.com.qmth.examcloud.core.basic.starter.config;
 
+import cn.com.qmth.examcloud.core.basic.service.report.RocketMqConsumerListener;
+import cn.com.qmth.examcloud.web.bootstrap.PropertyHolder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.boot.ApplicationArguments;
@@ -7,39 +9,23 @@ import org.springframework.boot.ApplicationRunner;
 import org.springframework.core.annotation.Order;
 import org.springframework.stereotype.Component;
 
-import cn.com.qmth.examcloud.core.basic.service.report.RocketMqConsumerListener;
-import cn.com.qmth.examcloud.reports.commons.enums.MqType;
-import cn.com.qmth.examcloud.web.bootstrap.PropertyHolder;
-
 @Component
 @Order(999)
 public class MqConsumerListenerStartup implements ApplicationRunner {
-	private static final Logger LOG = LoggerFactory.getLogger(MqConsumerListenerStartup.class);
-
-	private void startConsumerListener() {
-
-		String mqType = PropertyHolder.getString("$report.mq-type");
 
-		Boolean reportEnable = PropertyHolder.getBoolean("$report.enable", false);
+    private static final Logger log = LoggerFactory.getLogger(MqConsumerListenerStartup.class);
 
-		if (reportEnable) {
-			if(MqType.ROCKETMQ.getCode().equals(mqType)) {
-				RocketMqConsumerListener.start();
-			}else {
-				LOG.error("no $report.mq-type property config!");
-			}
-		}
+    private void startConsumerListener() {
+        Boolean reportEnable = PropertyHolder.getBoolean("examcloud.rocketmq.enable", false);
+        if (reportEnable) {
+            RocketMqConsumerListener.start();
+            log.warn("Rocketmq ConsumerListener started...");
+        }
+    }
 
-	}
+    @Override
+    public void run(ApplicationArguments args) throws Exception {
+        new Thread(() -> startConsumerListener()).start();
+    }
 
-	@Override
-	public void run(ApplicationArguments args) throws Exception {
-		new Thread() {
-			@Override
-			public void run() {
-				startConsumerListener();
-			}
-			
-		}.start();
-	}
 }