From c5aa48918abd3bce7ccc98b38476f0c22c6604c7 Mon Sep 17 00:00:00 2001 From: wxz Date: Wed, 14 Jul 2021 17:16:09 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E6=94=B9=EF=BC=9A=E6=8A=BD=E5=8F=96ro?= =?UTF-8?q?cketmq=E4=B8=BB=E8=A6=81=E6=B3=A8=E5=86=8C=E9=80=BB=E8=BE=91?= =?UTF-8?q?=E5=88=B0rocketmq=E9=80=9A=E7=94=A8=E6=A8=A1=E5=9D=97=EF=BC=8C?= =?UTF-8?q?=E5=90=84=E6=9C=8D=E5=8A=A1=E5=8F=AA=E5=81=9A=E6=9C=80=E7=AE=80?= =?UTF-8?q?=E5=8D=95=E7=9A=84=E8=B0=83=E7=94=A8=E5=8D=B3=E5=8F=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- epmet-commons/epmet-commons-rocketmq/pom.xml | 5 ++ .../register/ConsumerConfigProperties.java | 43 ---------- .../rocketmq/register/MQAbstractRegister.java | 77 ++++++++++++++++++ .../register/MQConsumerProperties.java | 29 +++++++ .../rocketmq/register/MQConsumerRegister.java | 79 ------------------- .../register/MQRegisterConfiguration.java | 35 ++++++++ .../epmet/mq/RocketMQConsumerRegister.java | 74 +++++------------ .../InitCustomerOrgRolesListener.java | 1 - .../src/main/resources/bootstrap.yml | 4 + 9 files changed, 171 insertions(+), 176 deletions(-) delete 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/MQAbstractRegister.java create mode 100644 epmet-commons/epmet-commons-rocketmq/src/main/java/com/epmet/commons/rocketmq/register/MQConsumerProperties.java delete mode 100644 epmet-commons/epmet-commons-rocketmq/src/main/java/com/epmet/commons/rocketmq/register/MQConsumerRegister.java create mode 100644 epmet-commons/epmet-commons-rocketmq/src/main/java/com/epmet/commons/rocketmq/register/MQRegisterConfiguration.java diff --git a/epmet-commons/epmet-commons-rocketmq/pom.xml b/epmet-commons/epmet-commons-rocketmq/pom.xml index 346f055ccc..75b7a9d1a5 100644 --- a/epmet-commons/epmet-commons-rocketmq/pom.xml +++ b/epmet-commons/epmet-commons-rocketmq/pom.xml @@ -26,5 +26,10 @@ org.projectlombok lombok + + com.epmet + epmet-commons-tools + 2.0.0 + 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 deleted file mode 100644 index 71a8fea41c..0000000000 --- a/epmet-commons/epmet-commons-rocketmq/src/main/java/com/epmet/commons/rocketmq/register/ConsumerConfigProperties.java +++ /dev/null @@ -1,43 +0,0 @@ -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/MQAbstractRegister.java b/epmet-commons/epmet-commons-rocketmq/src/main/java/com/epmet/commons/rocketmq/register/MQAbstractRegister.java new file mode 100644 index 0000000000..b1ca887c59 --- /dev/null +++ b/epmet-commons/epmet-commons-rocketmq/src/main/java/com/epmet/commons/rocketmq/register/MQAbstractRegister.java @@ -0,0 +1,77 @@ +package com.epmet.commons.rocketmq.register; + +import com.epmet.commons.tools.exception.ExceptionUtils; +import lombok.extern.slf4j.Slf4j; +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; + +/** + * @author wxz + * @Description 父类抽象注册器 + * @date 2021.07.14 15:38:21 + */ +@Slf4j +public abstract class MQAbstractRegister { + + /** + * @Description 注册所有监听器，由子类实现，做具体监听器的注册 + * @return + * @author wxz + * @date 2021.07.14 15:48 + */ + public abstract void registerAllListeners(String env, MQConsumerProperties consumerProperties); + + /** + * @Description 真正执行注册的方法，供子类直接调用，子类也可以覆盖该方法 + * @return + * @author wxz + * @date 2021.07.14 15:56 + */ + public void register(MQConsumerProperties consumerProperties, String group, MessageModel messageModel, String topic, String subExpression, MessageListener listener) { + try { + String nameServer = consumerProperties.getNameServer(); + Integer consumeThreadMin = consumerProperties.getConsumeThreadMin(); + Integer consumeThreadMax = consumerProperties.getConsumeThreadMax(); + String instanceName = buildInstanceName(); + // 实例化消费者 + DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group); + + // 设置NameServer的地址 + consumer.setNamesrvAddr(nameServer); + consumer.setMessageModel(messageModel); + consumer.setInstanceName(instanceName); + // 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息 + consumer.subscribe(topic, subExpression); + // 注册回调实现类来处理从broker拉取回来的消息 + consumer.registerMessageListener(listener); + if (consumeThreadMin != null) { + consumer.setConsumeThreadMin(consumeThreadMin); + } + if (consumeThreadMax != null) { + consumer.setConsumeThreadMax(consumeThreadMax); + } + // 启动消费者实例 + consumer.start(); + log.info(String.format("监听器注册完成,消费者组:%s,Topic:%s,Tag:%s,实例名称:%s", group, topic, subExpression, instanceName)); + } catch (Exception e) { + log.error(String.format("监听器注册失败,消费者组:%s,Topic:%s,Tag:%s。详细信息:%s", group, topic, subExpression, ExceptionUtils.getErrorStackTrace(e))); + } + } + + /** + * @Description 构造实例名称 + * @return + * @author wxz + * @date 2021.07.14 15:56 + */ + 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-commons/epmet-commons-rocketmq/src/main/java/com/epmet/commons/rocketmq/register/MQConsumerProperties.java b/epmet-commons/epmet-commons-rocketmq/src/main/java/com/epmet/commons/rocketmq/register/MQConsumerProperties.java new file mode 100644 index 0000000000..76af70e0e2 --- /dev/null +++ b/epmet-commons/epmet-commons-rocketmq/src/main/java/com/epmet/commons/rocketmq/register/MQConsumerProperties.java @@ -0,0 +1,29 @@ +package com.epmet.commons.rocketmq.register; + +import lombok.Data; +import org.springframework.boot.context.properties.ConfigurationProperties; + +/** + * @Description 注册属性 + * @author wxz + * @date 2021.07.14 15:33:16 +*/ +@Data +@ConfigurationProperties(prefix = "rocketmq") +public class MQConsumerProperties { + + /** + * nameServer + */ + private String nameServer; + + /** + * 最小消费线程数 + */ + private Integer consumeThreadMin; + /** + * 最大消费线程数 + */ + private Integer consumeThreadMax; + +} 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 deleted file mode 100644 index 3387b55931..0000000000 --- a/epmet-commons/epmet-commons-rocketmq/src/main/java/com/epmet/commons/rocketmq/register/MQConsumerRegister.java +++ /dev/null @@ -1,79 +0,0 @@ -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-commons/epmet-commons-rocketmq/src/main/java/com/epmet/commons/rocketmq/register/MQRegisterConfiguration.java b/epmet-commons/epmet-commons-rocketmq/src/main/java/com/epmet/commons/rocketmq/register/MQRegisterConfiguration.java new file mode 100644 index 0000000000..77e3d178f7 --- /dev/null +++ b/epmet-commons/epmet-commons-rocketmq/src/main/java/com/epmet/commons/rocketmq/register/MQRegisterConfiguration.java @@ -0,0 +1,35 @@ +package com.epmet.commons.rocketmq.register; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.annotation.Configuration; + +import javax.annotation.PostConstruct; + +/** + * @author wxz + * @Description MQ注册配置类 + * @date 2021.07.14 15:36:24 + */ +@Configuration +@ConditionalOnProperty(prefix = "rocketmq", name = "enable", havingValue = "true", matchIfMissing = false) +@EnableConfigurationProperties(MQConsumerProperties.class) +@Slf4j +public class MQRegisterConfiguration { + + @Value("${spring.profiles.active}") + private String env; + @Autowired + private MQConsumerProperties consumerProperties; + @Autowired + private MQAbstractRegister mqRegister; + + @PostConstruct + public void register() { + mqRegister.registerAllListeners(env, consumerProperties); + log.info("监听器注册动作执行完毕"); + } +} 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..c607cf5929 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 @@ -2,65 +2,33 @@ package com.epmet.mq; import com.epmet.commons.rocketmq.constants.ConsomerGroupConstants; import com.epmet.commons.rocketmq.constants.TopicConstants; +import com.epmet.commons.rocketmq.register.MQAbstractRegister; +import com.epmet.commons.rocketmq.register.MQConsumerProperties; import com.epmet.commons.tools.enums.EnvEnum; 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; - +/** + * @Description 如果rocketmq.enable=true，这里必须实现，且 实例化 + * @author wxz + * @date 2021.07.14 17:13:41 +*/ @Component -public class RocketMQConsumerRegister { - @Value("${spring.profiles.active}") - private String env; - @Value("${rocketmq.name-server}") - private String nameServer; - - /** - * @return - * @Description 注册监听器 - * @author wxz - * @date 2021.03.03 16:09 - */ - @PostConstruct - public void registerAllListeners() { - try { - if (!EnvEnum.LOCAL.getCode().equals(env)) { - 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 + ""); +public class RocketMQConsumerRegister extends MQAbstractRegister { + + @Override + public void registerAllListeners(String env, MQConsumerProperties consumerProperties) { + if (!EnvEnum.LOCAL.getCode().equals(env)) { + // 客户初始化监听器注册 + register(consumerProperties, + ConsomerGroupConstants.INIT_CUSTOMER_ORG_ROLES_GROUP, + MessageModel.CLUSTERING, + TopicConstants.INIT_CUSTOMER, + "*", + new InitCustomerOrgRolesListener()); + + // ...其他监听器类似 } - - return instanceName; } - } 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 010e2044a3..7224ea757a 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 @@ -28,7 +28,6 @@ import java.util.concurrent.TimeUnit; * @return * @date 2021.03.03 16:10 */ -//@Component public class InitCustomerOrgRolesListener implements MessageListenerConcurrently { private Logger logger = LoggerFactory.getLogger(getClass()); diff --git a/epmet-module/gov-org/gov-org-server/src/main/resources/bootstrap.yml b/epmet-module/gov-org/gov-org-server/src/main/resources/bootstrap.yml index 8baed16dd6..e7b5020a88 100644 --- a/epmet-module/gov-org/gov-org-server/src/main/resources/bootstrap.yml +++ b/epmet-module/gov-org/gov-org-server/src/main/resources/bootstrap.yml @@ -145,4 +145,8 @@ shutdown: waitTimeSecs: 30 # 优雅停机等待时间，超过30秒，发出告警 rocketmq: + # 是否开启mq + enable: true name-server: @rocketmq.nameserver@ +# consume-thread-min: 2 +# consume-thread-max: 2