统一告警注解和aop服务

liutongyi-admin/pom.xml

@@ -24,6 +24,18 @@
             <optional>true</optional> <!-- 表示依赖不会传递 -->
         </dependency>
 
+        <!--spring test-->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <scope>test</scope>
+        </dependency>
 
         <dependency>
             <groupId>com.dm</groupId>

liutongyi-admin/src/main/java/com/citygis/web/domain/KafkaPushTest.java

@@ -0,0 +1,20 @@
+package com.citygis.web.domain;
+
+import com.citygis.common.entity.KafkaEntity;
+import lombok.Data;
+
+/**
+ * @Name kafkapushtest
+ * @Version 1.0
+ * @Date 2025-02-21 15:10:09
+ * @Author harri
+ * @Description:
+ */
+
+@Data
+public class KafkaPushTest extends KafkaEntity {
+    // todo: instead by db object
+    private String querySql;
+
+    private int resCode = 0;
+}

liutongyi-admin/src/main/java/com/citygis/web/service/NotiKafkaService.java

@@ -0,0 +1,21 @@
+package com.citygis.web.service;
+
+import com.baomidou.mybatisplus.extension.service.IService;
+import com.citygis.common.annotation.PushToKafka;
+import com.citygis.web.domain.KafkaPushTest;
+
+/**
+ * @Name NotiKafkaService
+ * @Version 1.0
+ * @Date 2025-02-21 13:55:15
+ * @Author harri
+ * @Description: send to kafka service
+ */
+public interface NotiKafkaService {
+
+    /**
+     * get push params from monitoring register table
+     */
+    @PushToKafka()
+    public KafkaPushTest sendNotification(int monitorId);
+}

liutongyi-admin/src/main/java/com/citygis/web/service/impl/NotiKafkaServiceImpl.java

@@ -0,0 +1,31 @@
+package com.citygis.web.service.impl;
+
+import com.citygis.common.annotation.PushToKafka;
+import com.citygis.web.domain.KafkaPushTest;
+import com.citygis.web.service.NotiKafkaService;
+import org.springframework.stereotype.Service;
+
+/**
+ * @Name NotiKafkaServiceImpl
+ * @Version 1.0
+ * @Date 2025-02-21 15:16:33
+ * @Author harri
+ * @Description: Kafka Push Test
+ */
+
+@Service
+public class NotiKafkaServiceImpl implements NotiKafkaService {
+    // todo: 改为数据库查询结果
+    @Override
+    @PushToKafka
+    public KafkaPushTest sendNotification(int monitorId) {
+        KafkaPushTest kafkaPushTest = new KafkaPushTest();
+        kafkaPushTest.setQuerySql("dasd");
+        kafkaPushTest.setResCode(12);
+        kafkaPushTest.setBootstrapAddress("127.0.0.1");
+        kafkaPushTest.setPort("8080");
+        kafkaPushTest.setKey("asd");
+        kafkaPushTest.setValue("sssdadadasdasd");
+        return kafkaPushTest;
+    }
+}

liutongyi-common/src/main/java/com/citygis/common/annotation/PushToKafka.java

@@ -0,0 +1,47 @@
+package com.citygis.common.annotation;
+
+import java.lang.annotation.*;
+
+/**
+ * @Name PushToKafka
+ * @Version 1.0
+ * @Date 2025-02-14 15:52:51
+ * @Author harri
+ * @Description: unificatied alarm to kafka
+ */
+
+@Target(ElementType.METHOD)
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+public @interface PushToKafka {
+
+    /**
+     * Kafka 主题名称
+     */
+    String topic() default "test";
+
+    /**
+     * Kafka 消息的键值
+     * 默认为空字符串，表示不使用键值
+     */
+    String key() default "";
+
+    /**
+     * Kafka 消息的值
+     * 默认为空字符串，表示不使用默认值，需要通过方法参数传递
+     */
+    String value() default "";
+
+    /**
+     * Kafka server ip address
+     * 默认为 127.0.0.1，表示使用本地
+     */
+    String host() default "127.0.0.1";
+
+    /**
+     * Kafka port
+     * 默认为 8080，表示使用默认端口
+     */
+    String port() default "8080";
+
+}

liutongyi-common/src/main/java/com/citygis/common/entity/KafkaEntity.java

@@ -0,0 +1,39 @@
+package com.citygis.common.entity;
+
+import lombok.Data;
+
+/**
+ * @Name KafkaEntity
+ * @Version 1.0
+ * @Date 2025-02-20 17:13:44
+ * @Author harri
+ * @Description: common kafka entity
+ */
+
+@Data
+public class KafkaEntity {
+    /**
+     * kafka topic
+     */
+    private String topic;
+
+    /**
+     * kafka ip
+     */
+    private String bootstrapAddress;
+
+    /**
+     * kafka port
+     */
+    private String port;
+
+    /**
+     * kafka key
+     */
+    private String key;
+
+    /**
+     * kafka value
+     */
+    private String value;
+}

liutongyi-framework/src/main/java/com/citygis/framework/aspectj/KafkaPushAspect.java

@@ -0,0 +1,122 @@
+package com.citygis.framework.aspectj;
+
+import com.citygis.common.annotation.PushToKafka;
+import com.citygis.common.entity.KafkaEntity;
+import com.citygis.common.utils.StringUtils;
+import io.netty.util.internal.StringUtil;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.aspectj.lang.ProceedingJoinPoint;
+import org.aspectj.lang.annotation.Around;
+import org.aspectj.lang.annotation.Aspect;
+import org.aspectj.lang.reflect.MethodSignature;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.stereotype.Component;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+@Aspect
+@Component
+public class KafkaPushAspect {
+
+    private static final Logger log = LoggerFactory.getLogger(KafkaPushAspect.class);
+
+    private KafkaTemplate<String, String> kafkaTemplate;
+
+    @Around("@annotation(pushToKafka)")
+    public void doPushToKafka(ProceedingJoinPoint joinPoint, PushToKafka pushToKafka) throws Throwable {
+        System.out.println("dddddd");
+        try {
+            // 调用处理函数，将执行结果作为参数传递
+            handleSendMessage(joinPoint.proceed(), pushToKafka);
+        } catch (Exception e) {
+            // 处理异常
+            log.error("Error occurred during method execution: {}", e.getMessage());
+            throw e;
+        }
+    }
+
+    /**
+     * 处理发送消息到Kafka执行流程
+     * @param result 切点
+     * @param pushToKafka 注解
+     */
+    protected void handleSendMessage(final Object result, PushToKafka pushToKafka) {
+        // 校验目标方法传参的数据类型
+        if (!(result instanceof KafkaEntity)) {
+            log.warn("Invalid result type: {}", result.getClass().getName());
+            return;
+        }
+
+        String kafkaHost = replaceValidation(pushToKafka.host(), ((KafkaEntity) result).getBootstrapAddress());
+        String kafkaPort = replaceValidation(pushToKafka.port(), ((KafkaEntity) result).getPort());
+        String kafkaTopic = replaceValidation(pushToKafka.topic(), ((KafkaEntity) result).getTopic());
+        String kafkaKey = replaceValidation(pushToKafka.key(), ((KafkaEntity) result).getKey());
+        String kafkaValue = replaceValidation(pushToKafka.value(), ((KafkaEntity) result).getValue());
+
+        // kafka参数异常处理
+        if (
+                StringUtils.isEmpty(kafkaHost) || StringUtils.isEmpty(kafkaPort) ||
+                StringUtils.isEmpty(kafkaTopic) || StringUtils.isEmpty(kafkaKey) || StringUtils.isEmpty(kafkaValue)
+        ) {
+            log.warn("Invalid Kafka configuration: topic or host is empty");
+            return;
+        }
+
+        int kafkaPortInt;
+        try {
+            kafkaPortInt = Integer.parseInt(kafkaPort);
+        } catch (NumberFormatException e) {
+            log.error("Invalid Kafka port format: {}", kafkaPort);
+            return;
+        }
+
+        // 配置 KafkaTemplate
+        configureKafkaTemplate(kafkaHost, kafkaPortInt);
+
+        // 推送消息到kafka
+        try {
+            kafkaTemplate.send(kafkaTopic, kafkaKey, kafkaValue);
+            log.info("Pushed to Kafka: Topic={}, Key={}, Value={}", kafkaTopic, kafkaKey, kafkaValue);
+        } catch (Exception e) {
+            log.error("Error occurred while pushing to Kafka: {}", e.getMessage());
+        }
+    }
+
+    /**
+     * 变量替换
+     * @param initialValue 原始值
+     * @param compareValue 比对值
+     * @return 解析后的值
+     */
+    private String replaceValidation(String initialValue, String compareValue) {
+        if (compareValue != null && !compareValue.isEmpty()) {
+            return compareValue;
+        }
+        return initialValue;
+    }
+
+    /**
+     * 配置KafkaTemplate
+     * @param kafkaHost Kafka主机地址
+     * @param kafkaPort Kafka端口号
+     */
+    private void configureKafkaTemplate(String kafkaHost, int kafkaPort) {
+        Map<String, Object> configProps = new HashMap<>();
+        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaHost + ":" + kafkaPort);
+        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+
+        ProducerFactory<String, String> producerFactory = new DefaultKafkaProducerFactory<>(configProps);
+        kafkaTemplate = new KafkaTemplate<>(producerFactory);
+    }
+
+}