Merge branch 'dev_mq_init_customer' into dev_temp

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/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,11 +16,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;
+
 /**
  * 监听初始化客户动作，为客户初始化角色列表
  */
@@ -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/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);
         }
     }
 }