修改：抽取rocketmq主要注册逻辑到rocketmq通用模块，各服务只做最简单的调用即可

epmet-commons/epmet-commons-rocketmq/pom.xml

@@ -26,5 +26,10 @@
             <groupId>org.projectlombok</groupId>
             <artifactId>lombok</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.epmet</groupId>
+            <artifactId>epmet-commons-tools</artifactId>
+            <version>2.0.0</version>
+        </dependency>
     </dependencies>
 </project>

epmet-commons/epmet-commons-rocketmq/src/main/java/com/epmet/commons/rocketmq/register/ConsumerConfigProperties.java

@@ -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;
-}

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;
+    }
+}

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;
+
+}

epmet-commons/epmet-commons-rocketmq/src/main/java/com/epmet/commons/rocketmq/register/MQConsumerRegister.java

@@ -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;
-    }
-}

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("监听器注册动作执行完毕");
+    }
+}

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;
-
-@Component
-public class RocketMQConsumerRegister {
-    @Value("${spring.profiles.active}")
-    private String env;
-    @Value("${rocketmq.name-server}")
-    private String nameServer;
-
 /**
-     * @return
+ * @Description 如果rocketmq.enable=true，这里必须实现，且 实例化
-     * @Description 注册监听器
  * @author wxz
-     * @date 2021.03.03 16:09
+ * @date 2021.07.14 17:13:41
 */
-    @PostConstruct
+@Component
-    public void registerAllListeners() {
+public class RocketMQConsumerRegister extends MQAbstractRegister {
-        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() {
+    @Override
-        String instanceName = "";
+    public void registerAllListeners(String env, MQConsumerProperties consumerProperties) {
-        for (int i = 0; i < 4; i++) {
+        if (!EnvEnum.LOCAL.getCode().equals(env)) {
-            int t = (int) (Math.random() * 10);
+            // 客户初始化监听器注册
-            instanceName = instanceName.concat(t + "");
+            register(consumerProperties,
+                    ConsomerGroupConstants.INIT_CUSTOMER_ORG_ROLES_GROUP,
+                    MessageModel.CLUSTERING,
+                    TopicConstants.INIT_CUSTOMER,
+                    "*",
+                    new InitCustomerOrgRolesListener());
+
+            // ...其他监听器类似
         }
-
-        return instanceName;
     }
-
 }

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());

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