Merge branch 'dev'

epmet-module/gov-issue/gov-issue-server/src/main/java/com/epmet/mq/RocketMQConsumerRegister.java

@@ -0,0 +1,62 @@
+package com.epmet.mq;
+
+import com.epmet.commons.rocketmq.constants.ConsomerGroupConstants;
+import com.epmet.commons.rocketmq.constants.TopicConstants;
+import com.epmet.mq.listener.IssueProjectCategoryTagInitListener;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.listener.MessageListener;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+
+@Component
+public class RocketMQConsumerRegister {
+
+    @Value("${rocketmq.name-server}")
+    private String nameServer;
+
+    /**
+     * @return
+     * @Description 注册监听器
+     * @author wxz
+     * @date 2021.03.03 16:09
+     */
+    @PostConstruct
+    public void registerAllListeners() {
+        try {
+            register(nameServer, ConsomerGroupConstants.ISSUE_PROJECT_CATEGORY_TAG, MessageModel.CLUSTERING, TopicConstants.INIT_CUSTOMER, "*", new IssueProjectCategoryTagInitListener());
+        } catch (MQClientException e) {
+            e.printStackTrace();
+        }
+    }
+
+    public void register(String nameServer, String group, MessageModel messageModel, String topic, String subException, MessageListener listener) throws MQClientException {
+        // 实例化消费者
+        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group);
+
+        // 设置NameServer的地址
+        consumer.setNamesrvAddr(nameServer);
+        consumer.setMessageModel(messageModel);
+        consumer.setInstanceName(buildInstanceName());
+        // 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息
+        consumer.subscribe(topic, subException);
+        // 注册回调实现类来处理从broker拉取回来的消息
+        consumer.registerMessageListener(listener);
+        // 启动消费者实例
+        consumer.start();
+    }
+
+    private String buildInstanceName() {
+        String instanceName = "";
+        for (int i = 0; i < 4; i++) {
+            int t = (int) (Math.random() * 10);
+            instanceName = instanceName.concat(t + "");
+        }
+
+        return instanceName;
+    }
+
+}

epmet-module/gov-issue/gov-issue-server/src/main/java/com/epmet/mq/listener/IssueProjectCategoryTagInitListener.java

@@ -1,35 +1,28 @@
 package com.epmet.mq.listener;
 
 import com.alibaba.fastjson.JSON;
-import com.epmet.commons.rocketmq.constants.ConsomerGroupConstants;
-import com.epmet.commons.rocketmq.constants.TopicConstants;
 import com.epmet.commons.rocketmq.messages.InitCustomerMQMsg;
 import com.epmet.commons.tools.distributedlock.DistributedLock;
 import com.epmet.commons.tools.exception.ExceptionUtils;
 import com.epmet.commons.tools.exception.RenException;
 import com.epmet.dto.form.CategoryTagInitFormDTO;
 import com.epmet.service.IssueProjectCategoryDictService;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
+import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
 import org.apache.rocketmq.common.message.MessageExt;
-import org.apache.rocketmq.spring.annotation.MessageModel;
-import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
-import org.apache.rocketmq.spring.core.RocketMQListener;
 import org.redisson.api.RLock;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Component;
 
+import java.util.List;
 import java.util.concurrent.TimeUnit;
 
 /**
  * 监听初始化客户动作，为客户初始化议题、项目的分类、标签数据
  */
-@RocketMQMessageListener(topic = TopicConstants.INIT_CUSTOMER,
-        consumerGroup = ConsomerGroupConstants.ISSUE_PROJECT_CATEGORY_TAG,
-        messageModel = MessageModel.CLUSTERING,
-        selectorExpression = "*")
-@Component
-public class IssueProjectCategoryTagInitListener implements RocketMQListener<MessageExt> {
+public class IssueProjectCategoryTagInitListener implements MessageListenerConcurrently {
 
     private Logger logger = LoggerFactory.getLogger(getClass());
 
@@ -40,9 +33,19 @@ public class IssueProjectCategoryTagInitListener implements RocketMQListener<Mes
     private DistributedLock distributedLock;
 
     @Override
-    public void onMessage(MessageExt messageExt) {
+    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
+        try {
+            msgs.forEach(msg -> consumeMessage(msg));
+        } catch (Exception e) {
+            logger.error(ExceptionUtils.getErrorStackTrace(e));
+            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
+        }
+        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+    }
+
+    public void consumeMessage(MessageExt messageExt) {
         String msg = new String(messageExt.getBody());
-        logger.info("初始化客户-初始化客户自定义信息-收到消息内容:{}", msg);
+        logger.info("初始化客户-初始化议题、项目的分类、标签数据-收到消息内容:{}", msg);
         InitCustomerMQMsg msgObj = JSON.parseObject(msg, InitCustomerMQMsg.class);
 
         CategoryTagInitFormDTO dto = new CategoryTagInitFormDTO();
@@ -50,16 +53,16 @@ public class IssueProjectCategoryTagInitListener implements RocketMQListener<Mes
 
         RLock lock = null;
         try {
-            lock = distributedLock.getLock(String.format("lock:init_customer_home:%s", msgObj.getCustomerId()),
+            lock = distributedLock.getLock(String.format("lock:init_customer_project_category:%s", msgObj.getCustomerId()),
                     30l, 30l, TimeUnit.SECONDS);
             issueProjectCategoryDictService.init(dto);
         } catch (RenException e) {
             // 如果是我们手动抛出的异常，说明在业务可控范围内。目前不需要MQ重试
-            logger.error("【RocketMQ】初始化客户组件失败:".concat(ExceptionUtils.getErrorStackTrace(e)));
+            logger.error("【RocketMQ】初始化议题、项目的分类、标签数据失败:".concat(ExceptionUtils.getErrorStackTrace(e)));
         } catch (Exception e) {
             // 不是我们自己抛出的异常，可以让MQ重试
-            logger.error("【RocketMQ】初始化客户组件失败:".concat(ExceptionUtils.getErrorStackTrace(e)));
-             throw e;
+            logger.error("【RocketMQ】初始化议题、项目的分类、标签数据失败:".concat(ExceptionUtils.getErrorStackTrace(e)));
+            throw e;
         } finally {
             distributedLock.unLock(lock);
         }