Merge branch 'dev_mq_init_customer' into dev

epmet-commons/epmet-commons-tools/src/main/java/com/epmet/commons/tools/distributedlock/DistributedLock.java

@@ -37,21 +37,21 @@ public class DistributedLock {
     }
 
     /**
-     * @Description
+     * 取锁
-     * @Param name 锁名
+     * @param name 锁key
-     * @Param leaseTime 持锁时间 单位：min
+     * @param leaseTime 持有锁的时长
-     * @Param waitTime 获取锁最长等待时间 单位：min
+     * @param waitTime 取锁等待时长
-     * @author zxc
+     * @param timeUnit 持锁时长的时间单位
-     * @date 2020/10/29 9:04 上午
+     * @return
      */
-    public RLock getLock(String name,Long leaseTime,Long waitTime){
+    public RLock getLock(String name, Long leaseTime, Long waitTime, TimeUnit timeUnit) {
         RLock lock = null;
-        if (StringUtils.isNotBlank(name) && leaseTime > 0 && waitTime > 0 ){
+        if (StringUtils.isNotBlank(name) && leaseTime > 0 && waitTime > 0) {
             lock = redissonClient.getLock(name);
             Boolean lockStatus;
             try {
-                lockStatus = lock.tryLock(waitTime,leaseTime,TimeUnit.MINUTES);
+                lockStatus = lock.tryLock(waitTime, leaseTime, timeUnit);
-                if (!lockStatus){
+                if (!lockStatus) {
                     throw new RenException("获取锁🔒失败了......");
                 }
             } catch (InterruptedException e) {
@@ -62,14 +62,14 @@ public class DistributedLock {
     }
 
     /**
-     * @Description  释放锁🔒
+     * @Description 释放锁🔒
      * @Param rLock
      * @author zxc
      * @date 2020/10/28 2:52 下午
      */
-    public void unLock(RLock rLock){
+    public void unLock(RLock rLock) {
-        if (null != rLock){
+        if (null != rLock) {
-            if (rLock.isHeldByCurrentThread()){
+            if (rLock.isHeldByCurrentThread()) {
                 rLock.unlock();
             }
         }

epmet-module/gov-org/gov-org-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.InitCustomerOrgRolesListener;
+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.INIT_CUSTOMER_ORG_ROLES_GROUP, MessageModel.CLUSTERING, TopicConstants.INIT_CUSTOMER, "*", new InitCustomerOrgRolesListener());
+        } 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-org/gov-org-server/src/main/java/com/epmet/mq/listener/InitCustomerOrgListener.java

@@ -4,6 +4,7 @@ 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.constant.UserWorkType;
@@ -15,19 +16,22 @@ 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.concurrent.TimeUnit;
+
 /**
  * 监听初始化客户动作，为客户初始化角色列表
  */
-@RocketMQMessageListener(topic = TopicConstants.INIT_CUSTOMER,
+//@RocketMQMessageListener(topic = TopicConstants.INIT_CUSTOMER,
-        consumerGroup = ConsomerGroupConstants.INIT_CUSTOMER_ORG_ROLES_GROUP,
+//        consumerGroup = ConsomerGroupConstants.INIT_CUSTOMER_ORG_ROLES_GROUP,
-        messageModel = MessageModel.CLUSTERING,
+//        messageModel = MessageModel.CLUSTERING,
-        selectorExpression = "*")
+//        selectorExpression = "*")
-@Component
+//@Component
 public class InitCustomerOrgListener implements RocketMQListener<MessageExt> {
 
     private Logger logger = LoggerFactory.getLogger(getClass());
@@ -35,12 +39,19 @@ public class InitCustomerOrgListener implements RocketMQListener<MessageExt> {
     @Autowired
     private AgencyService agencyService;
 
+    @Autowired
+    private DistributedLock distributedLock;
+
     @Override
     public void onMessage(MessageExt messageExt) {
         String msg = new String(messageExt.getBody());
         logger.info("初始化客户-初始化组织信息-收到消息内容:{}", msg);
         InitCustomerMQMsg msgObj = JSON.parseObject(msg, InitCustomerMQMsg.class);
+
+        RLock lock = null;
         try {
+            lock = distributedLock.getLock(String.format("lock:init_customer_org:%s", msgObj.getCustomerId()),
+                    30l, 30l, TimeUnit.SECONDS);
             agencyService.saveRootAgency(constructRootAndAgencyDTO(msgObj));
         } catch (RenException e) {
             // 如果是我们手动抛出的异常，说明在业务可控范围内。目前不需要MQ重试
@@ -49,6 +60,8 @@ public class InitCustomerOrgListener implements RocketMQListener<MessageExt> {
             // 不是我们自己抛出的异常，可以让MQ重试
             logger.error("【RocketMQ】初始化客户组织失败:".concat(ExceptionUtils.getErrorStackTrace(e)));
             throw e;
+        } finally {
+            distributedLock.unLock(lock);
         }
     }
 

epmet-module/gov-org/gov-org-server/src/main/java/com/epmet/mq/listener/InitCustomerOrgRolesListener.java

@@ -0,0 +1,101 @@
+package com.epmet.mq.listener;
+
+import com.alibaba.fastjson.JSON;
+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.commons.tools.utils.SpringContextUtils;
+import com.epmet.constant.UserWorkType;
+import com.epmet.dto.CustomerAgencyDTO;
+import com.epmet.dto.form.AddAgencyAndStaffFormDTO;
+import com.epmet.dto.form.AdminStaffFromDTO;
+import com.epmet.service.AgencyService;
+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.redisson.api.RLock;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @Description 创建客户-组织和角色监听器
+ * @return
+ * @author wxz
+ * @date 2021.03.03 16:10
+*/
+public class InitCustomerOrgRolesListener implements MessageListenerConcurrently {
+
+    private Logger logger = LoggerFactory.getLogger(getClass());
+
+    @Override
+    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
+        try {
+            msgs.forEach(msg -> consumeMessage(msg));
+        } catch (Exception e) {
+            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+        }
+        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+    }
+
+    private void consumeMessage(MessageExt messageExt) {
+        String msg = new String(messageExt.getBody());
+        logger.info("初始化客户-初始化组织信息-收到消息内容:{}", msg);
+        InitCustomerMQMsg msgObj = JSON.parseObject(msg, InitCustomerMQMsg.class);
+
+        DistributedLock distributedLock = null;
+        RLock lock = null;
+        try {
+            distributedLock = SpringContextUtils.getBean(DistributedLock.class);
+            lock = distributedLock.getLock(String.format("lock:init_customer_org:%s", msgObj.getCustomerId()),
+                    30l, 30l, TimeUnit.SECONDS);
+            SpringContextUtils.getBean(AgencyService.class).saveRootAgency(constructRootAndAgencyDTO(msgObj));
+        } catch (RenException e) {
+            // 如果是我们手动抛出的异常，说明在业务可控范围内。目前不需要MQ重试
+            logger.error("【RocketMQ】初始化客户组织失败:".concat(ExceptionUtils.getErrorStackTrace(e)));
+        } catch (Exception e) {
+            // 不是我们自己抛出的异常，可以让MQ重试
+            logger.error("【RocketMQ】初始化客户组织失败:".concat(ExceptionUtils.getErrorStackTrace(e)));
+            throw e;
+        } finally {
+            distributedLock.unLock(lock);
+        }
+    }
+
+    /**
+     * @Description 构造dto
+     * @return
+     * @author wxz
+     * @date 2021.01.06 15:03
+     */
+    private AddAgencyAndStaffFormDTO constructRootAndAgencyDTO(InitCustomerMQMsg msgObj) {
+        AddAgencyAndStaffFormDTO agencyAndStaff = new AddAgencyAndStaffFormDTO();
+        //客户组织信息
+        CustomerAgencyDTO agencyDTO = new CustomerAgencyDTO();
+        agencyDTO.setId(msgObj.getAgency().getAgencyId());
+        agencyDTO.setCustomerId(msgObj.getCustomerId());
+        agencyDTO.setOrganizationName(msgObj.getAgency().getOrganizationName());
+        agencyDTO.setLevel(msgObj.getAgency().getLevel());
+        agencyDTO.setAreaCode(msgObj.getAgency().getAreaCode());
+        agencyDTO.setProvince(msgObj.getAgency().getProvince());
+        agencyDTO.setCity(msgObj.getAgency().getCity());
+        agencyDTO.setDistrict(msgObj.getAgency().getDistrict());
+        agencyAndStaff.setAgencyDTO(agencyDTO);
+
+        //客户管理员信息
+        AdminStaffFromDTO staffSubmitFrom = new AdminStaffFromDTO();
+        staffSubmitFrom.setCustomerId(msgObj.getCustomerId());
+        staffSubmitFrom.setAgencyId(msgObj.getStaff().getAgencyId());
+        staffSubmitFrom.setGender(msgObj.getStaff().getGender());
+        staffSubmitFrom.setMobile(msgObj.getStaff().getMobile());
+        staffSubmitFrom.setName(msgObj.getStaff().getName());
+        staffSubmitFrom.setWorkType(UserWorkType.FULL_TIME);
+        agencyAndStaff.setStaffDTO(staffSubmitFrom);
+
+        return agencyAndStaff;
+    }
+}

epmet-module/oper-customize/oper-customize-server/src/main/java/com/epmet/mq/listener/InitCustomerCustomizeListener.java

@@ -4,6 +4,7 @@ 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.CustomerHomeDTO;
@@ -12,11 +13,14 @@ 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.concurrent.TimeUnit;
+
 /**
  * 监听初始化客户动作，为客户初始化角色列表
  */
@@ -32,6 +36,9 @@ public class InitCustomerCustomizeListener implements RocketMQListener<MessageEx
     @Autowired
     private CustomerHomeService customerHomeService;
 
+    @Autowired
+    private DistributedLock distributedLock;
+
     @Override
     public void onMessage(MessageExt messageExt) {
         String msg = new String(messageExt.getBody());
@@ -40,7 +47,11 @@ public class InitCustomerCustomizeListener implements RocketMQListener<MessageEx
 
         CustomerHomeDTO customerHomeDTO = new CustomerHomeDTO();
         customerHomeDTO.setCustomerId(msgObj.getCustomerId());
+
+        RLock lock = null;
         try {
+            lock = distributedLock.getLock(String.format("lock:init_customer_home:%s", msgObj.getCustomerId()),
+                    30l, 30l, TimeUnit.SECONDS);
             customerHomeService.init(customerHomeDTO);
         } catch (RenException e) {
             // 如果是我们手动抛出的异常，说明在业务可控范围内。目前不需要MQ重试
@@ -48,7 +59,9 @@ public class InitCustomerCustomizeListener implements RocketMQListener<MessageEx
         } catch (Exception e) {
             // 不是我们自己抛出的异常，可以让MQ重试
             logger.error("【RocketMQ】初始化客户组件失败:".concat(ExceptionUtils.getErrorStackTrace(e)));
-            throw e;
+             throw e;
+        } finally {
+            distributedLock.unLock(lock);
         }
     }
 }