diff --git a/epmet-admin/epmet-admin-server/pom.xml b/epmet-admin/epmet-admin-server/pom.xml index cd0e462c73..2cf945e9b6 100644 --- a/epmet-admin/epmet-admin-server/pom.xml +++ b/epmet-admin/epmet-admin-server/pom.xml @@ -136,6 +136,7 @@ false + true 192.168.1.140:9876;192.168.1.141:9876 epmet_message @@ -172,6 +173,7 @@ false + false 192.168.1.140:9876;192.168.1.141:9876 epmet_message @@ -205,6 +207,7 @@ true + true 192.168.10.161:9876 epmet_message @@ -238,6 +241,7 @@ true + true 192.168.11.187:9876;192.168.11.184:9876 epmet_message diff --git a/epmet-admin/epmet-admin-server/src/main/java/com/epmet/mq/listener/RocketMQConsumerRegister.java b/epmet-admin/epmet-admin-server/src/main/java/com/epmet/mq/listener/RocketMQConsumerRegister.java index 1399fe6a13..8c07d1279d 100644 --- a/epmet-admin/epmet-admin-server/src/main/java/com/epmet/mq/listener/RocketMQConsumerRegister.java +++ b/epmet-admin/epmet-admin-server/src/main/java/com/epmet/mq/listener/RocketMQConsumerRegister.java @@ -2,6 +2,8 @@ package com.epmet.mq.listener; 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.listener.AuthOperationLogListener; import com.epmet.mq.listener.listener.PointOperationLogListener; @@ -16,55 +18,15 @@ 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; +public class RocketMQConsumerRegister extends MQAbstractRegister { - /** - * @return - * @Description 注册监听器 - * @author wxz - * @date 2021.03.03 16:09 - */ - @PostConstruct - public void registerAllListeners() { - try { - if (!EnvEnum.LOCAL.getCode().equals(env)) { - register(nameServer, ConsomerGroupConstants.AUTH_OPERATION_LOG_GROUP, MessageModel.CLUSTERING, TopicConstants.AUTH, "*", new AuthOperationLogListener()); - register(nameServer, ConsomerGroupConstants.PROJECT_OPERATION_LOG_GROUP, MessageModel.CLUSTERING, TopicConstants.PROJECT_CHANGED, "*", new ProjectOperationLogListener()); - register(nameServer, ConsomerGroupConstants.POINT_OPERATION_LOG_GROUP, MessageModel.CLUSTERING, TopicConstants.POINT, "*", new PointOperationLogListener()); - } - } catch (MQClientException e) { - e.printStackTrace(); - } - } - - public void register(String nameServer, String group, MessageModel messageModel, String topic, String subException, MessageListenerConcurrently 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 + ""); - } + @Override + public void registerAllListeners(String env, MQConsumerProperties consumerProperties) { + // 客户初始化监听器注册 + register(consumerProperties, ConsomerGroupConstants.AUTH_OPERATION_LOG_GROUP, MessageModel.CLUSTERING, TopicConstants.AUTH, "*", new AuthOperationLogListener()); + register(consumerProperties, ConsomerGroupConstants.PROJECT_OPERATION_LOG_GROUP, MessageModel.CLUSTERING, TopicConstants.PROJECT_CHANGED, "*", new ProjectOperationLogListener()); + register(consumerProperties, ConsomerGroupConstants.POINT_OPERATION_LOG_GROUP, MessageModel.CLUSTERING, TopicConstants.POINT, "*", new PointOperationLogListener()); - return instanceName; + // ...其他监听器类似 } - -} +} \ No newline at end of file diff --git a/epmet-admin/epmet-admin-server/src/main/resources/bootstrap.yml b/epmet-admin/epmet-admin-server/src/main/resources/bootstrap.yml index 0d9082cec3..74c6bc2567 100644 --- a/epmet-admin/epmet-admin-server/src/main/resources/bootstrap.yml +++ b/epmet-admin/epmet-admin-server/src/main/resources/bootstrap.yml @@ -136,4 +136,5 @@ shutdown: waitTimeSecs: 30 # 优雅停机等待时间，超过30秒，发出告警 rocketmq: + enable: @rocketmq.enable@ name-server: @rocketmq.nameserver@ \ No newline at end of file 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/data-statistical/data-statistical-server/pom.xml b/epmet-module/data-statistical/data-statistical-server/pom.xml index 7078d85f28..060a44c451 100644 --- a/epmet-module/data-statistical/data-statistical-server/pom.xml +++ b/epmet-module/data-statistical/data-statistical-server/pom.xml @@ -256,6 +256,7 @@ SECfcc020bdc83bb17a2c00f39977b1fbc409ef4188c7beaea11c5caa90eeaf87fd + true 192.168.1.140:9876;192.168.1.141:9876 @@ -375,6 +376,7 @@ SECfcc020bdc83bb17a2c00f39977b1fbc409ef4188c7beaea11c5caa90eeaf87fd + local 192.168.1.140:9876;192.168.1.141:9876 @@ -493,6 +495,7 @@ SECfcc020bdc83bb17a2c00f39977b1fbc409ef4188c7beaea11c5caa90eeaf87fd + true 192.168.10.161:9876 @@ -609,6 +612,7 @@ SEC95f4f40b533ad379ea6a6d1af6dd37029383cfe1b7cd96dfac2678be2c1c3ed1 + true 192.168.11.187:9876;192.168.11.184:9876 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 c383ae1b41..33ee4218e3 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 @@ -2,6 +2,8 @@ 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.constant.NumConstant; import com.epmet.commons.tools.enums.EnvEnum; import lombok.extern.slf4j.Slf4j; @@ -14,57 +16,13 @@ import org.springframework.stereotype.Component; import javax.annotation.PostConstruct; -@Slf4j @Component -public class RocketMQConsumerRegister { - @Value("${spring.profiles.active}") - private String env; - @Value("${rocketmq.name-server}") - private String nameServer; +public class RocketMQConsumerRegister extends MQAbstractRegister { - /** - * @return - * @Description 注册监听器 - * @author wxz - * @date 2021.03.03 16:09 - */ - @PostConstruct - public void registerAllListeners() { - try { - if (!EnvEnum.LOCAL.getCode().equals(env)) { - register(ConsomerGroupConstants.PROJECT_CHANGED_COMPONENTS_GROUP, MessageModel.CLUSTERING, TopicConstants.PROJECT_CHANGED, "*", new ProjectChangedCustomListener()); - } - } catch (MQClientException e) { - log.error("registerAllListeners exception", e); - } + @Override + public void registerAllListeners(String env, MQConsumerProperties consumerProperties) { + // 客户初始化监听器注册 + register(consumerProperties, ConsomerGroupConstants.PROJECT_CHANGED_COMPONENTS_GROUP, MessageModel.CLUSTERING, TopicConstants.PROJECT_CHANGED, "*", new ProjectChangedCustomListener()); + // ...其他监听器类似 } - - public void register(String group, MessageModel messageModel, String topic, String subException, MessageListenerConcurrently 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); - //一次批量拉去10条消息 - consumer.setConsumeMessageBatchMaxSize(NumConstant.TEN); - // 启动消费者实例 - 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; - } - -} +} \ No newline at end of file diff --git a/epmet-module/data-statistical/data-statistical-server/src/main/resources/bootstrap.yml b/epmet-module/data-statistical/data-statistical-server/src/main/resources/bootstrap.yml index c0479d72a3..7723414608 100644 --- a/epmet-module/data-statistical/data-statistical-server/src/main/resources/bootstrap.yml +++ b/epmet-module/data-statistical/data-statistical-server/src/main/resources/bootstrap.yml @@ -207,4 +207,5 @@ shutdown: waitTimeSecs: 30 # 优雅停机等待时间，超过30秒，发出告警 rocketmq: + enable: @rocketmq.enable@ name-server: @rocketmq.nameserver@ diff --git a/epmet-module/gov-issue/gov-issue-server/pom.xml b/epmet-module/gov-issue/gov-issue-server/pom.xml index 22bf17bb8e..ff1aa71829 100644 --- a/epmet-module/gov-issue/gov-issue-server/pom.xml +++ b/epmet-module/gov-issue/gov-issue-server/pom.xml @@ -175,6 +175,7 @@ SECfcc020bdc83bb17a2c00f39977b1fbc409ef4188c7beaea11c5caa90eeaf87fd + true 192.168.1.140:9876;192.168.1.141:9876 @@ -215,6 +216,7 @@ SECfcc020bdc83bb17a2c00f39977b1fbc409ef4188c7beaea11c5caa90eeaf87fd + false 192.168.1.140:9876;192.168.1.141:9876 @@ -254,6 +256,7 @@ https://oapi.dingtalk.com/robot/send?access_token=e894e5690f9d6a527722974c71548ff6c0fe29bd956589a09e21b16442a35ed4 SECfcc020bdc83bb17a2c00f39977b1fbc409ef4188c7beaea11c5caa90eeaf87fd + true 192.168.10.161:9876 @@ -294,6 +297,7 @@ + true 192.168.11.187:9876;192.168.11.184:9876 diff --git a/epmet-module/gov-issue/gov-issue-server/src/main/java/com/epmet/mq/RocketMQConsumerRegister.java b/epmet-module/gov-issue/gov-issue-server/src/main/java/com/epmet/mq/RocketMQConsumerRegister.java index 1674e82553..ff16d20238 100644 --- a/epmet-module/gov-issue/gov-issue-server/src/main/java/com/epmet/mq/RocketMQConsumerRegister.java +++ b/epmet-module/gov-issue/gov-issue-server/src/main/java/com/epmet/mq/RocketMQConsumerRegister.java @@ -2,6 +2,8 @@ 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.mq.listener.IssueProjectCategoryTagInitListener; import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; import org.apache.rocketmq.client.consumer.listener.MessageListener; @@ -13,50 +15,13 @@ import org.springframework.stereotype.Component; import javax.annotation.PostConstruct; @Component -public class RocketMQConsumerRegister { +public class RocketMQConsumerRegister extends MQAbstractRegister { - @Value("${rocketmq.name-server}") - private String nameServer; + @Override + public void registerAllListeners(String env, MQConsumerProperties consumerProperties) { + // 客户初始化监听器注册 + register(consumerProperties, ConsomerGroupConstants.ISSUE_PROJECT_CATEGORY_TAG, MessageModel.CLUSTERING, TopicConstants.INIT_CUSTOMER, "*", new IssueProjectCategoryTagInitListener()); - /** - * @return - * @Description 注册监听器 - * @author wxz - * @date 2021.03.03 16:09 - */ - @PostConstruct - public void registerAllListeners() { - try { - register(nameServer, ConsomerGroupConstants.ISSUE_PROJECT_CATEGORY_TAG, MessageModel.CLUSTERING, TopicConstants.INIT_CUSTOMER, "*", new IssueProjectCategoryTagInitListener()); - } 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; - } - -} +} \ No newline at end of file diff --git a/epmet-module/gov-issue/gov-issue-server/src/main/resources/bootstrap.yml b/epmet-module/gov-issue/gov-issue-server/src/main/resources/bootstrap.yml index 9e787eb546..09823befef 100644 --- a/epmet-module/gov-issue/gov-issue-server/src/main/resources/bootstrap.yml +++ b/epmet-module/gov-issue/gov-issue-server/src/main/resources/bootstrap.yml @@ -129,4 +129,5 @@ shutdown: waitTimeSecs: 30 # 优雅停机等待时间，超过30秒，发出告警 rocketmq: + enable: @rocketmq.enable@ name-server: @rocketmq.nameserver@ diff --git a/epmet-module/gov-org/gov-org-server/pom.xml b/epmet-module/gov-org/gov-org-server/pom.xml index 3c63b58119..c6dba17041 100644 --- a/epmet-module/gov-org/gov-org-server/pom.xml +++ b/epmet-module/gov-org/gov-org-server/pom.xml @@ -170,6 +170,7 @@ + true 192.168.1.140:9876;192.168.1.141:9876 epmet_message @@ -212,6 +213,7 @@ + false 192.168.1.140:9876;192.168.1.141:9876 epmet_message @@ -255,6 +257,7 @@ + true 192.168.10.161:9876 epmet_message @@ -295,6 +298,7 @@ + true 192.168.11.187:9876;192.168.11.184:9876 epmet_message 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..ea2474714b 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,36 @@ 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(); +public class RocketMQConsumerRegister extends MQAbstractRegister { + + @Override + public void registerAllListeners(String env, MQConsumerProperties consumerProperties) { + //判断逻辑，放到了了Pom.xml中，local不开启即可 + //if (!EnvEnum.LOCAL.getCode().equals(env)) { + //... + //} + + // 客户初始化监听器注册 + register(consumerProperties, + ConsomerGroupConstants.INIT_CUSTOMER_ORG_ROLES_GROUP, + MessageModel.CLUSTERING, + TopicConstants.INIT_CUSTOMER, + "*", + new InitCustomerOrgRolesListener()); + + // ...其他监听器类似 } - - 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/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..324ed0a5ca 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: @rocketmq.enable@ name-server: @rocketmq.nameserver@ +# consume-thread-min: 2 +# consume-thread-max: 2 diff --git a/epmet-module/gov-project/gov-project-client/src/main/java/com/epmet/dto/form/LatestListFormDTO.java b/epmet-module/gov-project/gov-project-client/src/main/java/com/epmet/dto/form/LatestListFormDTO.java index ca84aee3b0..c7f174086d 100644 --- a/epmet-module/gov-project/gov-project-client/src/main/java/com/epmet/dto/form/LatestListFormDTO.java +++ b/epmet-module/gov-project/gov-project-client/src/main/java/com/epmet/dto/form/LatestListFormDTO.java @@ -37,5 +37,7 @@ public class LatestListFormDTO implements Serializable { */ private Integer pageSize = 3; + private Integer num = 0; + } diff --git a/epmet-module/gov-project/gov-project-server/src/main/java/com/epmet/service/impl/ProjectServiceImpl.java b/epmet-module/gov-project/gov-project-server/src/main/java/com/epmet/service/impl/ProjectServiceImpl.java index 953104da7b..d721ccad51 100644 --- a/epmet-module/gov-project/gov-project-server/src/main/java/com/epmet/service/impl/ProjectServiceImpl.java +++ b/epmet-module/gov-project/gov-project-server/src/main/java/com/epmet/service/impl/ProjectServiceImpl.java @@ -1014,7 +1014,8 @@ public class ProjectServiceImpl extends BaseServiceImpl getClosedProjectList(LatestListFormDTO formDTO) { - + int pageIndex = (formDTO.getPageNo() - NumConstant.ONE) * formDTO.getPageSize(); + formDTO.setNum(pageIndex); return baseDao.selectClosedProjectList(formDTO); } diff --git a/epmet-module/gov-project/gov-project-server/src/main/resources/mapper/ProjectDao.xml b/epmet-module/gov-project/gov-project-server/src/main/resources/mapper/ProjectDao.xml index 11eeea95b4..a75b3514af 100644 --- a/epmet-module/gov-project/gov-project-server/src/main/resources/mapper/ProjectDao.xml +++ b/epmet-module/gov-project/gov-project-server/src/main/resources/mapper/ProjectDao.xml @@ -167,7 +167,8 @@ AND p.customer_id = #{customerId} AND p.`status` = 'closed' AND pro.operation = 'close' - ORDER BY p.created_time DESC + ORDER BY pro.created_time DESC + LIMIT #{num}, #{pageSize}