Browse Source

批量创建报警采用消息队列处理

lixing 3 years ago
parent
commit
96cb23ab8c

+ 2 - 0
dmp-alarm/src/main/java/com/persagy/dmp/alarm/enumeration/EnumAlarmMessageType.java

@@ -19,6 +19,8 @@ public enum EnumAlarmMessageType {
     ALARM_ITEM_DELETE("alarmItemDelete", "删除报警条目"),
     ALARM_RECORD_CREATE("alarmRecordCreate", "新增报警"),
     ALARM_RECORD_UPDATE("alarmRecordUpdate", "报警更新"),
+    ALARM_CONFIG_BATCH_CREATE("alarmConfigBatchCreate", "报警定义批量创建"),
+    ALARM_CONFIG_BATCH_DELETE("alarmConfigBatchDelete", "报警定义批量删除"),
     INSTANCE_OBJ_DELETE("instanceObjectDelete", "设备删除");
 
     @Getter

+ 2 - 25
dmp-alarm/src/main/java/com/persagy/dmp/alarm/jms/ConsumeMessageThreadPool.java

@@ -1,7 +1,6 @@
 package com.persagy.dmp.alarm.jms;
 
 import cn.hutool.core.thread.ExecutorBuilder;
-import com.persagy.common.json.JacksonMapper;
 import com.persagy.dmp.rwd.model.DmpMessage;
 import com.rabbitmq.client.Channel;
 import lombok.extern.slf4j.Slf4j;
@@ -36,31 +35,9 @@ public class ConsumeMessageThreadPool {
                 .build();
     }
 
-    public void consumeMessage(DmpMessage message, Channel channel, Message msg) {
+    public void consumeMessage(DmpMessage message) throws Exception {
         es.execute(() -> {
-            try {
-                log.info("============================== Receive:" + JacksonMapper.toSimpleJson(message));
-                messageProcesser.listen(message);
-                // 手动确认消息已消费
-                channel.basicAck(msg.getMessageProperties().getDeliveryTag(), false);
-            } catch (Exception e) {
-                log.error(e.getMessage(),e);
-                Boolean isRedeliveredFail = msg.getMessageProperties().getRedelivered();
-                // true表示消息已经重复处理失败
-                try {
-                    if (isRedeliveredFail) {
-                        // 拒绝消息,requeue=false 表示不再重新入队,如果配置了死信队列则进入死信队列
-                        log.error("重复消费消息失败,message: {}", JacksonMapper.toSimpleJson(message));
-                        channel.basicReject(msg.getMessageProperties().getDeliveryTag(), false);
-                    } else {
-                        // 如果是第一次失败则再次放入队列
-                        // requeue为是否重新回到队列,true重新入队
-                        channel.basicNack(msg.getMessageProperties().getDeliveryTag(), false, true);
-                    }
-                } catch (Exception exception) {
-                    log.error("重复消费消息失败", exception);
-                }
-            }
+            messageProcesser.listen(message);
         });
     }
 }

+ 11 - 4
dmp-alarm/src/main/java/com/persagy/dmp/alarm/jms/JmsConfig.java

@@ -17,6 +17,7 @@ import org.springframework.context.annotation.Configuration;
  * @verison: 1.0
  */
 @Slf4j
+@Deprecated
 //@Configuration
 public class JmsConfig {
     @Autowired
@@ -28,6 +29,12 @@ public class JmsConfig {
     @Value("${persagy.dmp.alarm.routingKey}")
     private String dmpAlarmRoutingKey;
 
+    /**
+     * 报警定义路由键
+     */
+    @Value("${persagy.dmp.alarm.config.routingKey}")
+    private String dmpAlarmConfigRoutingKey;
+
     @Value("${alarm.obj.routingKey}")
     private String alarmObjRoutingKey;
 
@@ -114,8 +121,8 @@ public class JmsConfig {
 //        consumeMessageThreadPool.consumeMessage(message, channel, msg);
 //    }
 
-    @RabbitListener(queues = "${persagy.dmp.alarm.queue}")    //监听器监听指定的Queue
-    public void processAlarmMsg(DmpMessage message, Channel channel, Message msg) {
-        consumeMessageThreadPool.consumeMessage(message, channel, msg);
-    }
+//    @RabbitListener(queues = "${persagy.dmp.alarm.queue}")    //监听器监听指定的Queue
+//    public void processAlarmMsg(DmpMessage message, Channel channel, Message msg) {
+//        consumeMessageThreadPool.consumeMessage(message, channel, msg);
+//    }
 }

+ 1 - 1
dmp-alarm/src/main/java/com/persagy/dmp/alarm/jms/MessageExecutor.java

@@ -9,6 +9,6 @@ import com.persagy.dmp.rwd.model.DmpMessage;
  */
 public interface MessageExecutor {
 
-	void execute(DmpMessage message);
+	void execute(DmpMessage message) throws Exception;
 
 }

+ 7 - 8
dmp-alarm/src/main/java/com/persagy/dmp/alarm/jms/MessageProcesser.java

@@ -42,14 +42,13 @@ public class MessageProcesser {
 	private static final Set<String> beanNames = new HashSet<>();
 
 	static {
-		beanNames.add(EnumAlarmMessageType.ALARM_TARGET_CREATE.getValue());
-		beanNames.add(EnumAlarmMessageType.ALARM_TARGET_UPDATE.getValue());
-		beanNames.add(EnumAlarmMessageType.ALARM_ITEM_UPDATE.getValue());
-		beanNames.add(EnumAlarmMessageType.INSTANCE_OBJ_DELETE.getValue());
-		beanNames.add(EnumAlarmMessageType.ALARM_TARGET_DELETE.getValue());
-//		由其他服务消费
-//		beanNames.add(RabbitMQConstants.ALARM_CONFIG_CREATE);
-//		beanNames.add(RabbitMQConstants.ALARM_CONFIG_DELETE);
+//		beanNames.add(EnumAlarmMessageType.ALARM_TARGET_CREATE.getValue());
+//		beanNames.add(EnumAlarmMessageType.ALARM_TARGET_UPDATE.getValue());
+//		beanNames.add(EnumAlarmMessageType.ALARM_ITEM_UPDATE.getValue());
+//		beanNames.add(EnumAlarmMessageType.INSTANCE_OBJ_DELETE.getValue());
+//		beanNames.add(EnumAlarmMessageType.ALARM_TARGET_DELETE.getValue());
+		beanNames.add(EnumAlarmMessageType.ALARM_CONFIG_BATCH_CREATE.getValue());
+//		beanNames.add(EnumAlarmMessageType.ALARM_CONFIG_BATCH_DELETE.getValue());
 	}
 
 	public void listen(DmpMessage message) {

+ 88 - 0
dmp-alarm/src/main/java/com/persagy/dmp/alarm/jms/RabbitmqConfig.java

@@ -0,0 +1,88 @@
+package com.persagy.dmp.alarm.jms;
+
+import com.persagy.dmp.rwd.model.DmpMessage;
+import com.rabbitmq.client.Channel;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.amqp.core.*;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * rabbitmq配置
+ *
+ * @author lixing
+ * @version V1.0 2021/10/11 7:32 下午
+ */
+@Slf4j
+@Configuration
+public class RabbitmqConfig {
+    @Autowired
+    ConsumeMessageThreadPool consumeMessageThreadPool;
+
+    /**
+     * 报警交换机
+     */
+    private final String exchange = "exchange-dmp";
+
+    /**
+     * 报警定义路由键
+     */
+    private final String dmpAlarmConfigRoutingKey = "alarm-config-routing-key";
+
+    /**
+     * 报警对象队列
+     */
+    private final String alarmConfigQueue = "alarm-config-queue";
+
+    @Bean
+    public TopicExchange exchange() {
+        return new TopicExchange(exchange);
+    }
+
+    @Bean
+    public Queue alarmConfigQueue() {
+        return new Queue(alarmConfigQueue, true);
+    }
+
+    @Bean
+    public Binding alarmObjBinding() {
+        return BindingBuilder.bind(alarmConfigQueue()).to(exchange()).with(dmpAlarmConfigRoutingKey);
+    }
+
+    /**
+     * 监听消息
+     *
+     * @param message DmpMessage(中台报警消息对象)
+     * @param channel channel对象,用于反馈消息消费结果
+     * @param msg rabbitmq消息对象
+     * @author lixing
+     * @version V1.0 2021/10/11 7:40 下午
+     */
+    @RabbitListener(queues = {alarmConfigQueue})
+    public void process(DmpMessage message, Channel channel, Message msg) throws Exception{
+        // 手动确认消息已消费
+        try {
+            log.info("============================== Receive:" + message);
+            if (message != null) {
+                consumeMessageThreadPool.consumeMessage(message);
+            }
+
+            channel.basicAck(msg.getMessageProperties().getDeliveryTag(), false);
+        } catch (Exception e) {
+            Boolean isRedeliveredFail = msg.getMessageProperties().getRedelivered();
+            // true表示消息已经重复处理失败
+            if (isRedeliveredFail) {
+                // 拒绝消息,requeue=false 表示不再重新入队,如果配置了死信队列则进入死信队列
+                log.error("重复消费消息失败,message: {}", message);
+                channel.basicReject(msg.getMessageProperties().getDeliveryTag(), false);
+            } else {
+                // 如果是第一次失败则再次放入队列
+                // requeue为是否重新回到队列,true重新入队
+                channel.basicNack(msg.getMessageProperties().getDeliveryTag(), false, true);
+            }
+        }
+    }
+}

+ 66 - 0
dmp-alarm/src/main/java/com/persagy/dmp/alarm/jms/executor/AlarmConfigBatchCreateExecutor.java

@@ -0,0 +1,66 @@
+package com.persagy.dmp.alarm.jms.executor;
+
+import com.alibaba.fastjson.JSON;
+import com.persagy.common.json.JacksonMapper;
+import com.persagy.common.web.BaseResponse;
+import com.persagy.common.web.MapResponse;
+import com.persagy.dmp.alarm.jms.MessageExecutor;
+import com.persagy.dmp.alarm.model.AlarmConfigModel;
+import com.persagy.dmp.alarm.service.AlarmConfigService;
+import com.persagy.dmp.alarm.util.CheckRequiredParam;
+import com.persagy.dmp.config.DmpParameterStorage;
+import com.persagy.dmp.rwd.model.DmpMessage;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+import org.springframework.util.CollectionUtils;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * 批量创建报警定义
+ *
+ * @author lixing
+ * @version V1.0 2021/10/11 8:00 下午
+ */
+@Slf4j
+@Component
+public class AlarmConfigBatchCreateExecutor implements MessageExecutor {
+    @Autowired
+    private AlarmConfigService alarmConfigService;
+
+    @Override
+    public void execute(DmpMessage message) throws Exception {
+        /* 验证消息是否符合要求 */
+        CheckRequiredParam.check(message, "exts");
+
+        Map<String, Object> exts = message.getExts();
+        String key = "alarmConfigs";
+        Object obj = exts.get(key);
+        List<AlarmConfigModel> configModels = new ArrayList<>();
+        if (obj instanceof ArrayList<?>) {
+            for (Object o : (List<?>) obj) {
+                AlarmConfigModel alarmConfigModel = JacksonMapper.toObject(
+                        JSON.toJSONString(o), AlarmConfigModel.class);
+                configModels.add(alarmConfigModel);
+            }
+        }
+        if (CollectionUtils.isEmpty(configModels)) {
+            return;
+        }
+        /* 设置通用请求参数 */
+        DmpParameterStorage.ParameterData data = new DmpParameterStorage.ParameterData();
+        data.setAppId(message.getAppId());
+        data.setGroupCode(message.getGroupCode());
+        data.setProjectId(message.getProjectId());
+        data.setUserId(message.getUserId());
+        DmpParameterStorage.set(data);
+
+        MapResponse response = alarmConfigService.batchCreate(configModels);
+        if (BaseResponse.FAIL.equals(response.getResult())) {
+            log.error("消费AlarmConfigBatchCreate消息异常,{}", response.getMessage());
+        }
+    }
+}

+ 13 - 19
dmp-alarm/src/main/java/com/persagy/dmp/alarm/service/AlarmConfigService.java

@@ -710,44 +710,38 @@ public class AlarmConfigService extends AlarmBaseService {
     }
 
     /**
-     * @description: 批量创建报警定义
-     * @param: configModels
-     * @return: com.persagy.common.web.MapResponse
-     * @exception:
-     * @author: lixing
-     * @company: Persagy Technology Co.,Ltd
-     * @since: 2020/12/4 6:45 下午
-     * @version: V1.0
+     * 批量创建报警定义
+     *
+     * @param configModels 报警定义列表
+     * @return 报警定义id列表
+     * @author lixing
+     * @version V1.0 2021/10/11 8:08 下午
      */
-    @Transactional
     public MapResponse batchCreate(List<AlarmConfigModel> configModels) {
         MapResponse response = new MapResponse();
         if (configModels == null) {
             response.setFail("请传入要创建的报警定义!");
             return response;
         }
-        // 一次最多更新200条数据
-        if (configModels.size() > 200) {
-            response.setFail("一次最多创建200条数据");
-            return response;
-        }
 
         AlarmParam alarmParam = prepareParam(response);
         if (alarmParam == null) {
             return response;
         }
         Date date = new Date();
-        List<String> ids = new ArrayList<>();
+        List<AlarmConfig> configs = new ArrayList<>(configModels.size());
         for (AlarmConfigModel configModel : configModels) {
             AlarmConfig entity = initAlarmConfigEntity(configModel, alarmParam, response);
             if (entity == null) {
+                response.setFail("批量创建报警条件失败,参数中存在null对象");
                 return response;
             }
-            alarmConfigRepository.save(entity);
-            ids.add(entity.getId());
-            DmpMessage msg = generateMessage(Lists.newArrayList(entity), null);
-            messageProcesser.convertAndSend(msg);
+            configs.add(entity);
         }
+        configs = alarmConfigRepository.saveAll(configs);
+        DmpMessage msg = generateMessage(configs, null);
+        messageProcesser.convertAndSend(msg);
+        List<String> ids = configs.stream().map(AlarmConfig::getId).collect(Collectors.toList());
         Date date1 = new Date();
         System.out.println("执行时间:" + (date1.getTime() - date.getTime()));
         response.add("ids", ids);

+ 10 - 5
dmp-alarm/src/main/resources/bootstrap.yml

@@ -5,10 +5,15 @@ spring:
     active: dev
   cloud:
     config:
-#      profile: aliyun-dev
-#      uri: http://39.102.43.179:9932
-      profile: dev2
-      uri: http://192.168.64.18:9932
+      profile: prod
+      uri: http://192.168.100.102:32435
+  jpa:
+    properties:
+      hibernate:
+        jdbc:
+          batch_size: 20
+        order_inserts: true
+        generate_statistics: true
 persagy:
   log:
     path: ../logs/dmp-alarm/
@@ -20,4 +25,4 @@ management:
         include: health
   endpoint:
     health:
-      enabled: true
+      enabled: true

+ 14 - 0
dmp-alarm/src/main/resources/logback-spring.xml

@@ -6,6 +6,7 @@
     <property name="app_name" value="alarm"/>
     <springProperty scope="context" name="log_path" source="persagy.log.path"/>
     <property name="log_file" value="${log_path}/${app_name}"/>
+    <property name="rabbit_log_file" value="${log_path}/${app_name}/rabbit/"/>
     <contextName>${app_name}</contextName>
 
     <appender name="stdout" class="ch.qos.logback.core.ConsoleAppender">
@@ -25,10 +26,23 @@
         </encoder>
     </appender>
 
+    <!-- 消息队列消息消费失败日志文件 -->
+    <appender name="rabbitmq" class="ch.qos.logback.core.rolling.RollingFileAppender">
+        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
+            <fileNamePattern>${rabbit_log_file}%d{yyyyMMdd}.log</fileNamePattern>
+        </rollingPolicy>
+        <encoder>
+            <pattern>${log_pattern}</pattern>
+        </encoder>
+    </appender>
+
     <logger name="org.hibernate" level="INFO"/>
     <logger name="org.springframework" level="INFO"/>
     <logger name="com.netflix" level="WARN"/>
     <logger name="org.apache.http" level="INFO"/>
+    <logger name="com.persagy.dmp.alarm.jms.RabbitmqConfig" level="ERROR">
+        <appender-ref ref="rabbitmq"/>
+    </logger>
 
     <springProfile name="dev">
         <logger name="org.hibernate.SQL" level="DEBUG"/>