From 9093c08bb411cce7c7847ea63f0378f23317b516 Mon Sep 17 00:00:00 2001 From: wxz Date: Thu, 4 Mar 2021 10:30:51 +0800 Subject: [PATCH 1/4] =?UTF-8?q?=E4=BF=AE=E6=94=B9rocketmq=E7=9A=84?= =?UTF-8?q?=E6=B3=A8=E5=86=8C=E6=96=B9=E5=BC=8F=EF=BC=9Aspringboot?= =?UTF-8?q?=E6=B3=A8=E5=86=8C->=E6=89=8B=E5=8A=A8api=E6=B3=A8=E5=86=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../InitCustomerOrgRolesListener.java | 3 +- .../epmet/mq/RocketMQConsumerRegister.java | 62 +++++++++++++++++ .../InitCustomerComponentsListener.java | 68 +++++++++++++++++++ .../InitCustomerCustomizeListener.java | 11 +-- 4 files changed, 138 insertions(+), 6 deletions(-) create mode 100644 epmet-module/oper-customize/oper-customize-server/src/main/java/com/epmet/mq/RocketMQConsumerRegister.java create mode 100644 epmet-module/oper-customize/oper-customize-server/src/main/java/com/epmet/mq/listener/InitCustomerComponentsListener.java 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 index 1225da9abd..97a15d7e56 100644 --- 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 @@ -37,7 +37,8 @@ public class InitCustomerOrgRolesListener implements MessageListenerConcurrently try { msgs.forEach(msg -> consumeMessage(msg)); } catch (Exception e) { - return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; + logger.error(ExceptionUtils.getErrorStackTrace(e)); + return ConsumeConcurrentlyStatus.RECONSUME_LATER; } return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } diff --git a/epmet-module/oper-customize/oper-customize-server/src/main/java/com/epmet/mq/RocketMQConsumerRegister.java b/epmet-module/oper-customize/oper-customize-server/src/main/java/com/epmet/mq/RocketMQConsumerRegister.java new file mode 100644 index 0000000000..057cf3c501 --- /dev/null +++ b/epmet-module/oper-customize/oper-customize-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.InitCustomerComponentsListener; +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_COMPONENTS_GROUP, MessageModel.CLUSTERING, TopicConstants.INIT_CUSTOMER, "*", new InitCustomerComponentsListener()); + } 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/oper-customize/oper-customize-server/src/main/java/com/epmet/mq/listener/InitCustomerComponentsListener.java b/epmet-module/oper-customize/oper-customize-server/src/main/java/com/epmet/mq/listener/InitCustomerComponentsListener.java new file mode 100644 index 0000000000..147d716674 --- /dev/null +++ b/epmet-module/oper-customize/oper-customize-server/src/main/java/com/epmet/mq/listener/InitCustomerComponentsListener.java @@ -0,0 +1,68 @@ +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.dto.CustomerHomeDTO; +import com.epmet.service.CustomerHomeService; +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; + +/** + * 监听初始化客户动作，为客户初始化角色列表 + */ +public class InitCustomerComponentsListener 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) { + logger.error(ExceptionUtils.getErrorStackTrace(e)); + return ConsumeConcurrentlyStatus.RECONSUME_LATER; + } + 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); + + CustomerHomeDTO customerHomeDTO = new CustomerHomeDTO(); + customerHomeDTO.setCustomerId(msgObj.getCustomerId()); + + DistributedLock distributedLock = null; + RLock lock = null; + try { + distributedLock = SpringContextUtils.getBean(DistributedLock.class); + lock = distributedLock.getLock(String.format("lock:init_customer_home:%s", msgObj.getCustomerId()), + 30l, 30l, TimeUnit.SECONDS); + + CustomerHomeService customerHomeService = SpringContextUtils.getBean(CustomerHomeService.class); + customerHomeService.init(customerHomeDTO); + } 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); + } + } +} \ No newline at end of file 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 6fa846c46c..7f7ddd5644 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 @@ -23,12 +23,13 @@ import java.util.concurrent.TimeUnit; /** * 监听初始化客户动作，为客户初始化角色列表 + * 已废弃 */ -@RocketMQMessageListener(topic = TopicConstants.INIT_CUSTOMER, - consumerGroup = ConsomerGroupConstants.INIT_CUSTOMER_COMPONENTS_GROUP, - messageModel = MessageModel.CLUSTERING, - selectorExpression = "*") -@Component +//@RocketMQMessageListener(topic = TopicConstants.INIT_CUSTOMER, +// consumerGroup = ConsomerGroupConstants.INIT_CUSTOMER_COMPONENTS_GROUP, +// messageModel = MessageModel.CLUSTERING, +// selectorExpression = "*") +//@Component public class InitCustomerCustomizeListener implements RocketMQListener { private Logger logger = LoggerFactory.getLogger(getClass()); From 4d4d17eed86dc58085445efb8425d38e58595614 Mon Sep 17 00:00:00 2001 From: jianjun Date: Fri, 30 Apr 2021 17:01:48 +0800 Subject: [PATCH 2/4] =?UTF-8?q?mq=E6=B3=A8=E5=86=8C=E4=BB=A3=E7=A0=81?= =?UTF-8?q?=E8=B0=83=E6=95=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../register/ConsumerConfigProperties.java | 43 +++++++++ .../rocketmq/register/MQConsumerRegister.java | 79 +++++++++++++++ .../mq/ProjectChangedCustomListener.java | 18 +++- .../epmet/mq/RocketMQConsumerRegister.java | 6 +- .../epmet/mq/RocketMQConsumerRegister.java | 3 +- .../mq/listener/InitCustomerOrgListener.java | 95 ------------------- .../InitCustomerOrgRolesListener.java | 24 ++++- .../GroupAchievementCustomListener.java | 22 ++++- .../mq/GroupAchieventConsumerListener.java | 36 ------- .../{type1 => }/RocketMQConsumerRegister.java | 8 +- 10 files changed, 186 insertions(+), 148 deletions(-) create mode 100644 epmet-commons/epmet-commons-rocketmq/src/main/java/com/epmet/commons/rocketmq/register/ConsumerConfigProperties.java create mode 100644 epmet-commons/epmet-commons-rocketmq/src/main/java/com/epmet/commons/rocketmq/register/MQConsumerRegister.java delete mode 100644 epmet-module/gov-org/gov-org-server/src/main/java/com/epmet/mq/listener/InitCustomerOrgListener.java rename epmet-module/resi-group/resi-group-server/src/main/java/com/epmet/mq/{type1 => }/GroupAchievementCustomListener.java (79%) delete mode 100644 epmet-module/resi-group/resi-group-server/src/main/java/com/epmet/mq/GroupAchieventConsumerListener.java rename epmet-module/resi-group/resi-group-server/src/main/java/com/epmet/mq/{type1 => }/RocketMQConsumerRegister.java (94%) diff --git a/epmet-commons/epmet-commons-rocketmq/src/main/java/com/epmet/commons/rocketmq/register/ConsumerConfigProperties.java b/epmet-commons/epmet-commons-rocketmq/src/main/java/com/epmet/commons/rocketmq/register/ConsumerConfigProperties.java new file mode 100644 index 0000000000..71a8fea41c --- /dev/null +++ b/epmet-commons/epmet-commons-rocketmq/src/main/java/com/epmet/commons/rocketmq/register/ConsumerConfigProperties.java @@ -0,0 +1,43 @@ +package com.epmet.commons.rocketmq.register; + +import lombok.Data; +import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; + +import java.io.Serializable; + +/** + * desc:mq 消费配置类 + * + * @author: LiuJanJun + * @date: 2021/4/30 2:39 下午 + * @version: 1.0 + */ +@Data +public class ConsumerConfigProperties implements Serializable { + + private static final long serialVersionUID = 2069676324708473773L; + /** + * 消费者组 + */ + private String consumerGroup; + /** + * 主题 + */ + private String topic; + /** + * 标签 + */ + private String tag = "*"; + /** + * 最小消费的线程数 + */ + private int consumeThreadMin = 2; + /** + * 最大消费的线程数 + */ + private int consumeThreadMax = 4; + /** + * 消费监听器 + */ + private MessageListenerConcurrently consumerListener; +} diff --git a/epmet-commons/epmet-commons-rocketmq/src/main/java/com/epmet/commons/rocketmq/register/MQConsumerRegister.java b/epmet-commons/epmet-commons-rocketmq/src/main/java/com/epmet/commons/rocketmq/register/MQConsumerRegister.java new file mode 100644 index 0000000000..f7e104f112 --- /dev/null +++ b/epmet-commons/epmet-commons-rocketmq/src/main/java/com/epmet/commons/rocketmq/register/MQConsumerRegister.java @@ -0,0 +1,79 @@ +package com.epmet.commons.rocketmq.register; + +import lombok.extern.slf4j.Slf4j; +import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; +import org.apache.rocketmq.client.exception.MQClientException; +import org.apache.rocketmq.common.protocol.heartbeat.MessageModel; +import org.springframework.beans.factory.annotation.Value; + +import javax.annotation.PostConstruct; + +/** + * desc：注册mq监听器 + * + * @author liujianjun + */ +@Slf4j +public abstract class MQConsumerRegister { + @Value("${spring.profiles.active}") + private String env; + @Value("${rocketmq.name-server}") + private String namesrvAddr; + + public abstract ConsumerConfigProperties getConsumerProperty(); + + + /** + * @return + * @Description 注册监听器 + * @author wxz + * @date 2021.03.03 16:09 + */ + @PostConstruct + public void registerMQListener() { + ConsumerConfigProperties consumerProperty = getConsumerProperty(); + log.info("registerAllListeners consumers:{} success", consumerProperty); + //本地环境不注册 + if ("local".equals(env)) { + try { + // 实例化消费者 + DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerProperty.getConsumerGroup()); + + // 设置NameServer的地址 + consumer.setNamesrvAddr(namesrvAddr); + consumer.setMessageModel(MessageModel.CLUSTERING); + consumer.setInstanceName(buildInstanceName()); + // 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息 + consumer.subscribe(consumer.getConsumerGroup(), consumerProperty.getTag()); + // 注册回调实现类来处理从broker拉取回来的消息 + consumer.registerMessageListener(consumerProperty.getConsumerListener()); + consumer.setConsumeThreadMin(consumerProperty.getConsumeThreadMin()); + consumer.setConsumeThreadMax(consumerProperty.getConsumeThreadMax()); + // 启动消费者实例 + consumer.start(); + } catch (MQClientException e) { + log.info("registerMQListener exception", e); + } + + } + + } + + /** + * desc: 因为docker-compose部署有问题 所有自己命名 + * + * @param + * @return java.lang.String + * @author LiuJanJun + * @date 2021/4/30 5:00 下午 + */ + 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/data-statistical/data-statistical-server/src/main/java/com/epmet/mq/ProjectChangedCustomListener.java b/epmet-module/data-statistical/data-statistical-server/src/main/java/com/epmet/mq/ProjectChangedCustomListener.java index 952d7b4ba5..1d51ddd077 100644 --- a/epmet-module/data-statistical/data-statistical-server/src/main/java/com/epmet/mq/ProjectChangedCustomListener.java +++ b/epmet-module/data-statistical/data-statistical-server/src/main/java/com/epmet/mq/ProjectChangedCustomListener.java @@ -1,7 +1,11 @@ package com.epmet.mq; 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.ProjectChangedMQMsg; +import com.epmet.commons.rocketmq.register.ConsumerConfigProperties; +import com.epmet.commons.rocketmq.register.MQConsumerRegister; import com.epmet.commons.tools.distributedlock.DistributedLock; import com.epmet.commons.tools.exception.RenException; import com.epmet.commons.tools.utils.SpringContextUtils; @@ -18,6 +22,7 @@ import org.apache.rocketmq.common.message.MessageExt; import org.redisson.api.RLock; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Component; import javax.annotation.PreDestroy; import java.util.Date; @@ -34,7 +39,8 @@ import java.util.stream.Collectors; * @date 2021.03.03 16:10 */ @Slf4j -public class ProjectChangedCustomListener implements MessageListenerConcurrently { +@Component +public class ProjectChangedCustomListener extends MQConsumerRegister implements MessageListenerConcurrently { private Logger logger = LoggerFactory.getLogger(getClass()); @@ -115,4 +121,14 @@ public class ProjectChangedCustomListener implements MessageListenerConcurrently log.info("data-statical-server服务被关闭，执行未执行完的动作"); } + + @Override + public ConsumerConfigProperties getConsumerProperty() { + ConsumerConfigProperties configProperties = new ConsumerConfigProperties(); + configProperties.setConsumerGroup(ConsomerGroupConstants.PROJECT_CHANGED_COMPONENTS_GROUP); + configProperties.setTopic(TopicConstants.PROJECT_CHANGED); + configProperties.setTag("*"); + configProperties.setConsumerListener(this); + return configProperties; + } } diff --git a/epmet-module/data-statistical/data-statistical-server/src/main/java/com/epmet/mq/RocketMQConsumerRegister.java b/epmet-module/data-statistical/data-statistical-server/src/main/java/com/epmet/mq/RocketMQConsumerRegister.java index ba6a515e6d..b0c6f8378a 100644 --- a/epmet-module/data-statistical/data-statistical-server/src/main/java/com/epmet/mq/RocketMQConsumerRegister.java +++ b/epmet-module/data-statistical/data-statistical-server/src/main/java/com/epmet/mq/RocketMQConsumerRegister.java @@ -10,14 +10,12 @@ import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; 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.core.annotation.Order; -import org.springframework.stereotype.Component; import javax.annotation.PostConstruct; @Slf4j -@Component -@Order(value = 111) +//@Component +//@Order(value = 111) public class RocketMQConsumerRegister { @Value("${spring.profiles.active}") private String env; 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 index bfe0f9f00f..c43db9b46d 100644 --- 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 @@ -9,11 +9,10 @@ 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 +//@Component public class RocketMQConsumerRegister { @Value("${spring.profiles.active}") private String env; 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 deleted file mode 100644 index c4012090b4..0000000000 --- a/epmet-module/gov-org/gov-org-server/src/main/java/com/epmet/mq/listener/InitCustomerOrgListener.java +++ /dev/null @@ -1,95 +0,0 @@ -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.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.common.message.MessageExt; -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 java.util.concurrent.TimeUnit; - -/** - * 监听初始化客户动作，为客户初始化角色列表 - */ -//@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()); - - @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重试 - 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/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 index ddd9bb8764..6ad695fc02 100644 --- 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 @@ -1,7 +1,11 @@ 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.rocketmq.register.ConsumerConfigProperties; +import com.epmet.commons.rocketmq.register.MQConsumerRegister; import com.epmet.commons.tools.distributedlock.DistributedLock; import com.epmet.commons.tools.exception.ExceptionUtils; import com.epmet.commons.tools.exception.RenException; @@ -18,17 +22,19 @@ import org.apache.rocketmq.common.message.MessageExt; import org.redisson.api.RLock; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Component; import java.util.List; import java.util.concurrent.TimeUnit; /** + * @author wxz * @Description 创建客户-组织和角色监听器 * @return - * @author wxz * @date 2021.03.03 16:10 -*/ -public class InitCustomerOrgRolesListener implements MessageListenerConcurrently { + */ +@Component +public class InitCustomerOrgRolesListener extends MQConsumerRegister implements MessageListenerConcurrently { private Logger logger = LoggerFactory.getLogger(getClass()); @@ -67,8 +73,8 @@ public class InitCustomerOrgRolesListener implements MessageListenerConcurrently } /** - * @Description 构造dto * @return + * @Description 构造dto * @author wxz * @date 2021.01.06 15:03 */ @@ -98,4 +104,14 @@ public class InitCustomerOrgRolesListener implements MessageListenerConcurrently return agencyAndStaff; } + + @Override + public ConsumerConfigProperties getConsumerProperty() { + ConsumerConfigProperties configProperties = new ConsumerConfigProperties(); + configProperties.setConsumerGroup(ConsomerGroupConstants.INIT_CUSTOMER_ORG_ROLES_GROUP); + configProperties.setTopic(TopicConstants.INIT_CUSTOMER); + configProperties.setTag("*"); + configProperties.setConsumerListener(this); + return configProperties; + } } diff --git a/epmet-module/resi-group/resi-group-server/src/main/java/com/epmet/mq/type1/GroupAchievementCustomListener.java b/epmet-module/resi-group/resi-group-server/src/main/java/com/epmet/mq/GroupAchievementCustomListener.java similarity index 79% rename from epmet-module/resi-group/resi-group-server/src/main/java/com/epmet/mq/type1/GroupAchievementCustomListener.java rename to epmet-module/resi-group/resi-group-server/src/main/java/com/epmet/mq/GroupAchievementCustomListener.java index cdd0ce615a..f3d98bd0d8 100644 --- a/epmet-module/resi-group/resi-group-server/src/main/java/com/epmet/mq/type1/GroupAchievementCustomListener.java +++ b/epmet-module/resi-group/resi-group-server/src/main/java/com/epmet/mq/GroupAchievementCustomListener.java @@ -1,7 +1,11 @@ -package com.epmet.mq.type1; +package com.epmet.mq; 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.GroupAchievementMQMsg; +import com.epmet.commons.rocketmq.register.ConsumerConfigProperties; +import com.epmet.commons.rocketmq.register.MQConsumerRegister; import com.epmet.commons.tools.distributedlock.DistributedLock; import com.epmet.commons.tools.exception.RenException; import com.epmet.commons.tools.utils.SpringContextUtils; @@ -15,6 +19,7 @@ import org.apache.rocketmq.common.message.MessageExt; import org.redisson.api.RLock; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Component; import java.util.List; import java.util.concurrent.TimeUnit; @@ -26,7 +31,8 @@ import java.util.concurrent.TimeUnit; * @date 2021.03.03 16:10 */ @Slf4j -public class GroupAchievementCustomListener implements MessageListenerConcurrently { +@Component +public class GroupAchievementCustomListener extends MQConsumerRegister implements MessageListenerConcurrently { private Logger logger = LoggerFactory.getLogger(getClass()); @@ -82,4 +88,16 @@ public class GroupAchievementCustomListener implements MessageListenerConcurrent } } } + + + + @Override + public ConsumerConfigProperties getConsumerProperty() { + ConsumerConfigProperties configProperties = new ConsumerConfigProperties(); + configProperties.setConsumerGroup(ConsomerGroupConstants.GROUP_ACHIEVEMENT_COMPONENTS_GROUP); + configProperties.setTopic(TopicConstants.GROUP_ACHIEVEMENT); + configProperties.setTag("*"); + configProperties.setConsumerListener(this); + return configProperties; + } } diff --git a/epmet-module/resi-group/resi-group-server/src/main/java/com/epmet/mq/GroupAchieventConsumerListener.java b/epmet-module/resi-group/resi-group-server/src/main/java/com/epmet/mq/GroupAchieventConsumerListener.java deleted file mode 100644 index 2f6d9594e9..0000000000 --- a/epmet-module/resi-group/resi-group-server/src/main/java/com/epmet/mq/GroupAchieventConsumerListener.java +++ /dev/null @@ -1,36 +0,0 @@ -package com.epmet.mq; - -import com.alibaba.fastjson.JSON; -import com.epmet.commons.rocketmq.messages.GroupAchievementMQMsg; -import com.epmet.modules.group.service.StatsAchievementService; -import lombok.extern.slf4j.Slf4j; -import org.apache.commons.lang3.StringUtils; -import org.apache.rocketmq.spring.core.RocketMQListener; -import org.springframework.beans.factory.annotation.Autowired; - -/** - * desc: 小组成就mq监听 - * - * @author: LiuJanJun - * @date: 2021/4/22 10:10 下午 - * @version: 1.0 - */ -@Slf4j -//@Component -//@RocketMQMessageListener(topic = "group_achievement", consumerGroup = ConsomerGroupConstants.GROUP_ACHIEVEMENT_COMPONENTS_GROUP) -public class GroupAchieventConsumerListener implements RocketMQListener { - @Autowired - private StatsAchievementService statsAchievementService; - - @Override - public void onMessage(GroupAchievementMQMsg msg) { - log.info("receive msg:{}", JSON.toJSONString(msg)); - if (StringUtils.isBlank(msg.getAchievementType()) || StringUtils.isBlank(msg.getCustomerId()) - || StringUtils.isBlank(msg.getGroupId())) { - log.error("consumer groupAchievement fail,msg:{}", msg); - return; - } - Boolean aBoolean = statsAchievementService.calculateAcm(msg.getCustomerId(), msg.getGroupId(), msg.getAchievementType()); - log.info("consumer groupAchievement msg success,{}", aBoolean); - } -} diff --git a/epmet-module/resi-group/resi-group-server/src/main/java/com/epmet/mq/type1/RocketMQConsumerRegister.java b/epmet-module/resi-group/resi-group-server/src/main/java/com/epmet/mq/RocketMQConsumerRegister.java similarity index 94% rename from epmet-module/resi-group/resi-group-server/src/main/java/com/epmet/mq/type1/RocketMQConsumerRegister.java rename to epmet-module/resi-group/resi-group-server/src/main/java/com/epmet/mq/RocketMQConsumerRegister.java index 3289dbdf2e..f11eaa28d7 100644 --- a/epmet-module/resi-group/resi-group-server/src/main/java/com/epmet/mq/type1/RocketMQConsumerRegister.java +++ b/epmet-module/resi-group/resi-group-server/src/main/java/com/epmet/mq/RocketMQConsumerRegister.java @@ -1,4 +1,4 @@ -package com.epmet.mq.type1; +package com.epmet.mq; import com.epmet.commons.rocketmq.constants.ConsomerGroupConstants; import com.epmet.commons.rocketmq.constants.TopicConstants; @@ -9,12 +9,11 @@ import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; 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; @Slf4j -@Component +//@Component public class RocketMQConsumerRegister { @Value("${spring.profiles.active}") private String env; @@ -30,7 +29,7 @@ public class RocketMQConsumerRegister { @PostConstruct public void registerAllListeners() { try { - if (!EnvEnum.LOCAL.getCode().equals(env)){ + if (!EnvEnum.LOCAL.getCode().equals(env)) { register(ConsomerGroupConstants.GROUP_ACHIEVEMENT_COMPONENTS_GROUP, MessageModel.CLUSTERING, TopicConstants.GROUP_ACHIEVEMENT, "*", new GroupAchievementCustomListener()); } } catch (MQClientException e) { @@ -48,6 +47,7 @@ public class RocketMQConsumerRegister { consumer.setInstanceName(buildInstanceName()); // 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息 consumer.subscribe(topic, subException); + consumer.setConsumeMessageBatchMaxSize(10); // 注册回调实现类来处理从broker拉取回来的消息 consumer.registerMessageListener(listener); // 启动消费者实例 From 3c88c7d35f11ffb4fb15a6693d0e9a6215f85f24 Mon Sep 17 00:00:00 2001 From: jianjun Date: Fri, 30 Apr 2021 17:24:57 +0800 Subject: [PATCH 3/4] =?UTF-8?q?mq=E6=B3=A8=E5=86=8C=E4=BB=A3=E7=A0=81?= =?UTF-8?q?=E8=B0=83=E6=95=B4-=E6=9C=AC=E5=9C=B0=E4=B8=8D=E6=B3=A8?= =?UTF-8?q?=E5=86=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../com/epmet/commons/rocketmq/register/MQConsumerRegister.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/epmet-commons/epmet-commons-rocketmq/src/main/java/com/epmet/commons/rocketmq/register/MQConsumerRegister.java b/epmet-commons/epmet-commons-rocketmq/src/main/java/com/epmet/commons/rocketmq/register/MQConsumerRegister.java index f7e104f112..4c883e74a7 100644 --- a/epmet-commons/epmet-commons-rocketmq/src/main/java/com/epmet/commons/rocketmq/register/MQConsumerRegister.java +++ b/epmet-commons/epmet-commons-rocketmq/src/main/java/com/epmet/commons/rocketmq/register/MQConsumerRegister.java @@ -34,7 +34,7 @@ public abstract class MQConsumerRegister { ConsumerConfigProperties consumerProperty = getConsumerProperty(); log.info("registerAllListeners consumers:{} success", consumerProperty); //本地环境不注册 - if ("local".equals(env)) { + if (!"local".equals(env)) { try { // 实例化消费者 DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerProperty.getConsumerGroup()); From 524a0c592ec0477ec551741d7104274dda501e66 Mon Sep 17 00:00:00 2001 From: jianjun Date: Fri, 30 Apr 2021 17:37:00 +0800 Subject: [PATCH 4/4] =?UTF-8?q?mq=E6=B3=A8=E5=86=8C=E4=BB=A3=E7=A0=81?= =?UTF-8?q?=E8=B0=83=E6=95=B4-=E6=BC=8F=E6=8E=89=E7=9A=84=E4=BB=A3?= =?UTF-8?q?=E7=A0=81=E5=90=88=E5=B9=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../epmet/mq/RocketMQConsumerRegister.java | 2 +- .../InitCustomerComponentsListener.java | 20 +++++- .../InitCustomerCustomizeListener.java | 68 ------------------- 3 files changed, 19 insertions(+), 71 deletions(-) delete mode 100644 epmet-module/oper-customize/oper-customize-server/src/main/java/com/epmet/mq/listener/InitCustomerCustomizeListener.java diff --git a/epmet-module/oper-customize/oper-customize-server/src/main/java/com/epmet/mq/RocketMQConsumerRegister.java b/epmet-module/oper-customize/oper-customize-server/src/main/java/com/epmet/mq/RocketMQConsumerRegister.java index 057cf3c501..c128db1207 100644 --- a/epmet-module/oper-customize/oper-customize-server/src/main/java/com/epmet/mq/RocketMQConsumerRegister.java +++ b/epmet-module/oper-customize/oper-customize-server/src/main/java/com/epmet/mq/RocketMQConsumerRegister.java @@ -12,7 +12,7 @@ import org.springframework.stereotype.Component; import javax.annotation.PostConstruct; -@Component +//@Component public class RocketMQConsumerRegister { @Value("${rocketmq.name-server}") diff --git a/epmet-module/oper-customize/oper-customize-server/src/main/java/com/epmet/mq/listener/InitCustomerComponentsListener.java b/epmet-module/oper-customize/oper-customize-server/src/main/java/com/epmet/mq/listener/InitCustomerComponentsListener.java index 147d716674..36fb8c930b 100644 --- a/epmet-module/oper-customize/oper-customize-server/src/main/java/com/epmet/mq/listener/InitCustomerComponentsListener.java +++ b/epmet-module/oper-customize/oper-customize-server/src/main/java/com/epmet/mq/listener/InitCustomerComponentsListener.java @@ -1,7 +1,11 @@ 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.rocketmq.register.ConsumerConfigProperties; +import com.epmet.commons.rocketmq.register.MQConsumerRegister; import com.epmet.commons.tools.distributedlock.DistributedLock; import com.epmet.commons.tools.exception.ExceptionUtils; import com.epmet.commons.tools.exception.RenException; @@ -15,6 +19,7 @@ import org.apache.rocketmq.common.message.MessageExt; import org.redisson.api.RLock; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Component; import java.util.List; import java.util.concurrent.TimeUnit; @@ -22,7 +27,8 @@ import java.util.concurrent.TimeUnit; /** * 监听初始化客户动作，为客户初始化角色列表 */ -public class InitCustomerComponentsListener implements MessageListenerConcurrently { +@Component +public class InitCustomerComponentsListener extends MQConsumerRegister implements MessageListenerConcurrently { private Logger logger = LoggerFactory.getLogger(getClass()); @@ -65,4 +71,14 @@ public class InitCustomerComponentsListener implements MessageListenerConcurrent distributedLock.unLock(lock); } } -} \ No newline at end of file + + @Override + public ConsumerConfigProperties getConsumerProperty() { + ConsumerConfigProperties configProperties = new ConsumerConfigProperties(); + configProperties.setConsumerGroup(ConsomerGroupConstants.INIT_CUSTOMER_COMPONENTS_GROUP); + configProperties.setTopic(TopicConstants.INIT_CUSTOMER); + configProperties.setTag("*"); + configProperties.setConsumerListener(this); + return configProperties; + } +} 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 deleted file mode 100644 index 7f7ddd5644..0000000000 --- a/epmet-module/oper-customize/oper-customize-server/src/main/java/com/epmet/mq/listener/InitCustomerCustomizeListener.java +++ /dev/null @@ -1,68 +0,0 @@ -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.CustomerHomeDTO; -import com.epmet.service.CustomerHomeService; -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_COMPONENTS_GROUP, -// messageModel = MessageModel.CLUSTERING, -// selectorExpression = "*") -//@Component -public class InitCustomerCustomizeListener implements RocketMQListener { - - private Logger logger = LoggerFactory.getLogger(getClass()); - - @Autowired - private CustomerHomeService customerHomeService; - - @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); - - 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重试 - logger.error("【RocketMQ】初始化客户组件失败:".concat(ExceptionUtils.getErrorStackTrace(e))); - } catch (Exception e) { - // 不是我们自己抛出的异常，可以让MQ重试 - logger.error("【RocketMQ】初始化客户组件失败:".concat(ExceptionUtils.getErrorStackTrace(e))); - throw e; - } finally { - distributedLock.unLock(lock); - } - } -} \ No newline at end of file