diff --git a/epmet-commons/epmet-commons-tools/src/main/java/com/epmet/commons/tools/distributedlock/DistributedLock.java b/epmet-commons/epmet-commons-tools/src/main/java/com/epmet/commons/tools/distributedlock/DistributedLock.java index 6bfb25fe48..85a2e61c84 100644 --- a/epmet-commons/epmet-commons-tools/src/main/java/com/epmet/commons/tools/distributedlock/DistributedLock.java +++ b/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 leaseTime 持锁时间 单位：min - * @Param waitTime 获取锁最长等待时间 单位：min - * @author zxc - * @date 2020/10/29 9:04 上午 + * 取锁 + * @param name 锁key + * @param leaseTime 持有锁的时长 + * @param waitTime 取锁等待时长 + * @param timeUnit 持锁时长的时间单位 + * @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); - if (!lockStatus){ + lockStatus = lock.tryLock(waitTime, leaseTime, timeUnit); + 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){ - if (null != rLock){ - if (rLock.isHeldByCurrentThread()){ + public void unLock(RLock rLock) { + if (null != rLock) { + if (rLock.isHeldByCurrentThread()) { rLock.unlock(); } } diff --git a/epmet-module/gov-org/gov-org-server/src/main/java/com/epmet/mq/RocketMQConsumerRegister.java b/epmet-module/gov-org/gov-org-server/src/main/java/com/epmet/mq/RocketMQConsumerRegister.java new file mode 100644 index 0000000000..b378946ff3 --- /dev/null +++ b/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; + } + +} diff --git a/epmet-module/gov-org/gov-org-server/src/main/java/com/epmet/mq/listener/InitCustomerOrgListener.java b/epmet-module/gov-org/gov-org-server/src/main/java/com/epmet/mq/listener/InitCustomerOrgListener.java index e5eb8dabd6..6ddc79801b 100644 --- a/epmet-module/gov-org/gov-org-server/src/main/java/com/epmet/mq/listener/InitCustomerOrgListener.java +++ b/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, - consumerGroup = ConsomerGroupConstants.INIT_CUSTOMER_ORG_ROLES_GROUP, - messageModel = MessageModel.CLUSTERING, - selectorExpression = "*") -@Component +//@RocketMQMessageListener(topic = TopicConstants.INIT_CUSTOMER, +// consumerGroup = ConsomerGroupConstants.INIT_CUSTOMER_ORG_ROLES_GROUP, +// messageModel = MessageModel.CLUSTERING, +// selectorExpression = "*") +//@Component public class InitCustomerOrgListener implements RocketMQListener { private Logger logger = LoggerFactory.getLogger(getClass()); @@ -35,12 +39,19 @@ public class InitCustomerOrgListener implements RocketMQListener { @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 { // 不是我们自己抛出的异常，可以让MQ重试 logger.error("【RocketMQ】初始化客户组织失败:".concat(ExceptionUtils.getErrorStackTrace(e))); throw e; + } finally { + distributedLock.unLock(lock); } } diff --git a/epmet-module/gov-org/gov-org-server/src/main/java/com/epmet/mq/listener/InitCustomerOrgRolesListener.java b/epmet-module/gov-org/gov-org-server/src/main/java/com/epmet/mq/listener/InitCustomerOrgRolesListener.java new file mode 100644 index 0000000000..1225da9abd --- /dev/null +++ b/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 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; + } +} diff --git a/epmet-module/oper-customize/oper-customize-server/src/main/java/com/epmet/mq/listener/InitCustomerCustomizeListener.java b/epmet-module/oper-customize/oper-customize-server/src/main/java/com/epmet/mq/listener/InitCustomerCustomizeListener.java index 882db74489..6fa846c46c 100644 --- a/epmet-module/oper-customize/oper-customize-server/src/main/java/com/epmet/mq/listener/InitCustomerCustomizeListener.java +++ b/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