[新增]一键报警从kafka取数据，并存入到数据库。提供业务系统使用

manailin

2021-09-27 26a67128b10df80f91367a92044ab36e119e8e0e

[新增]一键报警从kafka取数据，并存入到数据库。提供业务系统使用

 springcloud_k8s_panzhihuazhihuishequ/service_property/pom.xml

@@ -111,6 +111,23 @@
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>
        <!--kafka依赖 -->
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <version>2.7.6</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka-test</artifactId>
            <scope>test</scope>
            <version>2.7.6</version>
        </dependency>
        <!--json转换工具包依赖 -->
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
        </dependency>
    </dependencies>



 springcloud_k8s_panzhihuazhihuishequ/service_property/src/main/java/com/panzhihua/service_property/config/KafkaConfig.java

New file
@@ -0,0 +1,31 @@
package com.panzhihua.service_property.config;

import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.TopicBuilder;

/**
 * @author manailin
 * @version 1.0
 * @description kafka配置类 开启Kafka监听器标注的端点
 * @date 2021/9/26 15:33
 */
@Configuration
@EnableKafka
public class KafkaConfig {

    @Value("${kafka.topic.kafka-demo-topic}")
    private String orderTopic;

    /** 条件匹配，如果配置kafka.topic.autoCreate=true的话，该配置类生效，反之则无效 */
    @Bean
    @ConditionalOnProperty(name = "kafka.topic.autoCreate", havingValue = "true")
    public NewTopic orderTopic() {
        return TopicBuilder.name(orderTopic).build();
    }

}

 springcloud_k8s_panzhihuazhihuishequ/service_property/src/main/java/com/panzhihua/service_property/dto/CallForward.java

New file
@@ -0,0 +1,70 @@
package com.panzhihua.service_property.dto;

import lombok.Data;

/**
 * 应用模块名称
 * <p>
 * * 应用模块名称
 * <p>
 * 订阅一键报警的设备信息主题返回的对象 代码描述
 * <p>
 * Copyright; Copyright (C) 2021 XXX, Inc. All rights reserved.
 * <p>
 * Company; 成都呐喊信息技术有限公司
 * <p>
 *
 * @author manailin
 * @since 2021/9/26 16;33
 */
@Data
public class CallForward {
    /**
     * 呼叫转移记录GUID
     */
    private String CallForwardId;
    /**
     * 呼叫者GUID
     */
    private String Caller;
    /**
     * 呼叫顺序
     */
    private String Order;
    /**
     * 被呼叫者GUID
     */
    private String Callee;
    /**
     * 预定的呼叫时间
     */
    private String CallDateTime;
    /**
     * 呼叫次数
     */
    private String CallNums;
    /**
     * 呼叫持续时间
     */
    private String CallSeconds;
    /**
     * 呼叫策略
     */
    private String CallStrate;
    /**
     * 策略名称
     */
    private String ProName;
    /**
     * 是否自动接听
     */
    private String AutoAnswer;
    /**
     * 呼叫等待mp3文件id
     */
    private String WaitMp3Id;
    /**
     * 呼叫应答mp3文件id
     */
    private String CallMp3Id;
}

 springcloud_k8s_panzhihuazhihuishequ/service_property/src/main/java/com/panzhihua/service_property/dto/CallForwardDto.java

New file
@@ -0,0 +1,45 @@
package com.panzhihua.service_property.dto;

import lombok.Data;

/**
 * 应用模块名称
 * <p>
 * 订阅一键报警的设备信息主题返回的对象 代码描述
 * <p>
 * Copyright: Copyright (C) 2021 XXX, Inc. All rights reserved.
 * <p>
 * Company: 成都呐喊信息技术有限公司
 * <p>
 *
 * @author manailin
 * @since 2021/9/26 16:23
 */
@Data
public class CallForwardDto {

    private String CallForwardId;
    /** 呼叫转移记录GUID */
    private String Caller;
    /** 呼叫者GUID */
    private String Order;
    /** 呼叫顺序 */
    private String Callee;
    /** 被呼叫者GUID */
    private String CallDateTime;
    /** 预定的呼叫时间 */
    private String CallNums;
    /** 呼叫次数 */
    private String CallSeconds;
    /** 呼叫持续时间 */
    private String CallStratus;
    /** 呼叫策略 */
    private String ProName;
    /** 策略名称 */
    private String AutoAnswer;
    /** 是否自动接听 */
    private String WaitMp3Id;
    /** 呼叫等待mp3文件id */
    private String CallMp3Id;
    /** 呼叫应答mp3文件id */
}

 springcloud_k8s_panzhihuazhihuishequ/service_property/src/main/java/com/panzhihua/service_property/dto/CustomBtn.java

New file
@@ -0,0 +1,69 @@
package com.panzhihua.service_property.dto;

import lombok.Data;

/**
 * 应用模块名称
 * <p>
 * 代码描述
 * <p>
 * Copyright: Copyright (C) 2021 XXX, Inc. All rights reserved.
 * <p>
 * Company: 成都呐喊信息技术有限公司
 * <p>
 *
 * @author manailin
 * @since 2021/9/26 16:23
 */
@Data
public class CustomBtn {

    /**
     * 按键所属终端id
     */
    private String TerminalId;
    /**
     * 按键编码
     */
    private String BtnCode;
    /**
     * 按键任务guid
     */
    private String TskGuid;
    /**
     * 按键对应的响应终端id
     */
    private String ResTerminalId;
    /**
     * 播放曲目文件id
     */
    private String FileId;
    /**
     * 播放音量
     */
    private String SoundVol;
    /**
     * 灯光模式
     */
    private String LightMode;
    /**
     * 报警强切
     */
    private String CutAlarm;
    /**
     * 短路输出1
     */
    private String IoOut1;
    /**
     * 短路输出2
     */
    private String IoOut2;
    /**
     * 终端短路输出频率
     */
    private String IoOutFreQuency;
    /**
     * 终端短路输出持续时间
     */
    private String IoOutDuration;
}

 springcloud_k8s_panzhihuazhihuishequ/service_property/src/main/java/com/panzhihua/service_property/dto/DeviceInfoDto.java

New file
@@ -0,0 +1,255 @@
package com.panzhihua.service_property.dto;

import java.util.ArrayList;

import lombok.*;

/**
 * 应用模块名称
 * <p>
 * 订阅一键报警的设备信息主题返回的对象 代码描述
 * <p>
 * Copyright;Copyright (C) 2021 XXX Inc. All rights reserved.
 * <p>
 * Company;成都呐喊信息技术有限公司
 * <p>
 *
 * @author manailin
 * @since 2021/9/26 16:33
 */
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
@ToString
public class DeviceInfoDto {

    private String DeviceType;
    /**
     * 设备类型
     */
    private String DeviceName;
    /**
     * 设备用户名
     */
    private String DeviceNickName;
    /**
     * 设备用户昵称
     */
    private String DeviceGuid;
    /**
     * 设备GUID
     */
    private String DevicePwd;
    /**
     * 设备用户密码
     */
    private String DeviceAreaId;
    /**
     * 用户所在分区GUID
     */
    private String DeviceIp;
    /**
     * 设备IP
     */
    private String DevicePriority;
    /**
     * 设备用户级别
     */
    private String DeviceStatus;
    /**
     * 设备用户状态
     */
    private String HasVid;
    /**
     * 有无视频
     */
    private String HasTalk;
    /**
     * 有无对讲
     */
    private String RtspUrl;
    /**
     * 关联rtsp相机流地址
     */
    private String IoInCnt;
    /**
     * IO输入口数量
     */
    private String IoIn1State;
    /**
     * IO输入口1状态
     */
    private String IoIn2State;
    /**
     * IO输入口2状态
     */
    private String IoIn3State;
    /**
     * IO输入口3状态
     */
    private String IoIn4State;
    /**
     * IO输入口4状态
     */
    private String IoIn5State;
    /**
     * IO输入口5状态
     */
    private String IoIn6State;
    /**
     * IO输入口6状态
     */
    private String IoIn2Tp;
    /**
     * 报警输入2类型
     */
    private String IoOutCnt;
    /**
     * IO输出口数量
     */
    private String IoOut1State;
    /**
     * IO输出口1状态
     */
    private String IoOut2State;
    /**
     * IO输出口2状态
     */
    private String IoOut3State;
    /**
     * IO输出口3状态
     */
    private String IoOut4State;
    /**
     * IO输出口4状态
     */
    private String IoOut5State;
    /**
     * IO输出口5状态
     */
    private String IoOut6State;
    /**
     * IO输出口6状态
     */
    private String PlayTalkVol;
    /**
     * 对讲的播放音量
     */
    private String PlayBcVol;
    /**
     * 广播的播放音量
     */
    private String PlayBellVol;
    /**
     * 铃声的播放音量
     */
    private String CollTalkVol;
    /**
     * 采集的对讲音量
     */
    private String CollWatchVol;
    /**
     * 采集的监视音量
     */
    private String DeviceSerialNumber;
    /**
     * 设备序列号
     */
    private String ManufactureDate;
    /**
     * 出厂日期
     */
    private String DbAlarm;
    /**
     * 喧哗报警阈值
     */
    private String Model;
    /**
     * 设备模式
     */
    private String MoDet1;
    /**
     * 设备模式
     */
    private String MoDet2;
    /**
     * 设备模式
     */
    private String DeamonTime;
    /**
     * ARM守护程序发布时间
     */
    private String CompileTime;
    /**
     * ARM主程序发布时间
     */
    private String DeviceAbility;
    /**
     * 设备功能
     */
    private String DeviceCode;
    /**
     * 设备编码
     */
    private String DeviceDesc;
    /**
     * 设备描述
     */
    private String DeviceToken;
    /**
     * 加密狗id
     */
    private String TimeStamp;
    /**
     * 当前时间戳
     */
    private String DeviceVer;
    /**
     * 设备版本
     */
    private String CurrentTaskGuid;
    /**
     * 当前执行的任务guid
     */
    private String CurrentSoundSrc;
    /**
     * 当前音源
     */
    private String SipUsr;
    /**
     * sip账号
     */
    private String SipPwd;
    /**
     * sip密码
     */
    private String ProtocalType;
    /**
     * 协议类型 1:私有协议 2:sip协议 3:都有
     */
    private String TaskStatus;
    /**
     * 任务状态
     */
    private String CtlType;
    /**
     * 更改类型 1 添加 2 更新 3 删除 4权限改变
     */
    private String ReservedField1;
    /**
     * 扩展字段1
     */
    private String ReservedField2;
    /**
     * 扩展字段2
     */
    private String ReservedField3;
    /**
     * 扩展字段3
     */

    private ArrayList<CallForwardDto> CallForwardList;

    private ArrayList<CustomBtn> customBtns;
}

 springcloud_k8s_panzhihuazhihuishequ/service_property/src/main/java/com/panzhihua/service_property/dto/TaskDetail.java

New file
@@ -0,0 +1,130 @@
package com.panzhihua.service_property.dto;

import java.util.ArrayList;

import lombok.Data;

/**
 * 应用模块名称
 * <p>
 * 一键报警kafka 订阅主题 任务消息对象 代码描述
 * <p>
 * Copyright;Copyright (C) 2021 XXX Inc. All rights reserved.
 * <p>
 * Company;成都呐喊信息技术有限公司
 * <p>
 *
 * @author manailin
 * @since 2021/9/26 16:33
 */
@Data
public class TaskDetail {
    /**
     * 对讲服务地址
     */
    private String TalkServer;
    /**
     * 对讲服务端口
     */
    private String TalkPort;
    /**
     * 任务拨号时间
     */
    private String TskCallStartTime;
    /**
     * 任务启动时间
     */
    private String TskStartTime;
    /**
     * 任务guid
     */
    private String TskGuid;
    /**
     * 任务级别
     */
    private String TskPrio;
    /**
     * 呼叫者guid
     */
    private String CallerGuid;
    /**
     * 被呼叫者guid
     */
    private String CalleeGuid;
    /**
     * 呼叫者等级
     */
    private String CallerLevel;
    /**
     * 任务状态
     */
    private String TskStatus;
    /**
     * 任务参数：通道号
     */
    private String ChIdx;
    /**
     * 任务参数：传输方式
     */
    private String TransProto;
    /**
     * 任务参数：数据类型
     */
    private String MdTp;
    /**
     * 对讲音量
     */
    private String TalkVolumn;
    /**
     * 第三方厂家策略
     */
    private String CustomJson;
    /**
     * 呼叫转移记录GUID
     */
    private ArrayList<CallForward> CallForwardList;
    /**
     * 任务超时时间unix时间戳
     */
    private String TskTimeout;
    /**
     * 码流属性，0主码流 1子码流
     */
    private String StrmMode;
    /**
     * 是否已处理 1未 2已
     */
    private String AlreadyDeal;
    /**
     * 创建者guid
     */
    private String Creater;
    /**
     * 启动者guid
     */
    private String Starter;
    /**
     * 呼叫者名字
     */
    private String CallerName;
    /**
     * 被呼叫者名字
     */
    private String CalleeName;
    /**
     * 协议类型 1:私有协议 2:sip协议 3:都有
     */
    private String ProtocalType;
    /**
     * 扩展字段1 发起者类型 1 ARM 2 PC
     */
    private String ReservedField1;
    /**
     * 扩展字段2
     */
    private String ReservedField2;
    /**
     * 扩展字段3
     */
    private String ReservedField3;
}

 springcloud_k8s_panzhihuazhihuishequ/service_property/src/main/java/com/panzhihua/service_property/dto/TaskInfo.java

New file
@@ -0,0 +1,28 @@
package com.panzhihua.service_property.dto;

import lombok.Data;

/**
 * 应用模块名称
 * <p>
 * 一键报警kafka 订阅主题 任务消息对象 代码描述
 * <p>
 * Copyright: Copyright (C) 2021 XXX, Inc. All rights reserved.
 * <p>
 * Company: 成都呐喊信息技术有限公司
 * <p>
 *
 * @author manailin
 * @since 2021/9/26 16:30
 */
@Data
public class TaskInfo {

    /** 任务类型(1) */
    private Integer task_type;
    /** 更改类型 1 添加 2 更新 3 删除 */
    private Integer CtlType;
    /** 任务 */
    private TaskDetail talk;

}

 springcloud_k8s_panzhihuazhihuishequ/service_property/src/main/java/com/panzhihua/service_property/listener/OrderListener.java

New file
@@ -0,0 +1,54 @@
package com.panzhihua.service_property.listener;

import java.util.Date;

import javax.annotation.Resource;

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSONObject;
import com.panzhihua.service_property.dao.ComPropertyAlarmDao;
import com.panzhihua.service_property.dto.TaskInfo;
import com.panzhihua.service_property.entity.ComPropertyAlarm;

import lombok.extern.slf4j.Slf4j;

/**
 * @author manailin
 * @version 1.0
 * @description kafka 消息消费者
 * @date 2021/9/26 15:33
 */
@Slf4j
@Component
public class OrderListener {

    @Resource
    private ComPropertyAlarmDao comPropertyAlarmDao;

    /** 用于监听kafka，符合条件的消息能够即时消费 */
    @KafkaListener(topics = {"${kafka.topic.kafka-demo-topic}"}, autoStartup = "${kafka.topic.autoCreate:true}")
    public void processing(String message) throws Exception {
        log.info("监控来消息了：{}", message);
        try {
            TaskInfo task = JSONObject.parseObject(message, TaskInfo.class);
            if (task.getCtlType() == 1) {
                saveTask(task);
            }
        } catch (Exception e) {
            throw new Exception();
        }
    }

    @Async
    void saveTask(TaskInfo task) {
        ComPropertyAlarm comPropertyAlarm = new ComPropertyAlarm();
        comPropertyAlarm.setSerialNo(task.getTalk().getCallerGuid());
        comPropertyAlarm.setStatus(1);
        comPropertyAlarm.setCreateTime(new Date());
        comPropertyAlarmDao.insert(comPropertyAlarm);
    }

}

 springcloud_k8s_panzhihuazhihuishequ/service_property/src/main/resources/bootstrap.yml

@@ -1,6 +1,25 @@
spring:
  application:
    name: property
  kafka:
    bootstrap-servers: 182.151.37.143:9092 #kafka地址
    consumer:
      group-id: order-consumer
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer #kafkaTemplate.send()方法参数类型
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      auto-offset-reset: earliest
      security:
        protocol: PLAINTEXT
    producer:
      batch-size: 16384
      buffer-memory: 33554432
      retries: 0
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      properties:
        linger.ms: 1
      security:
        protocol: PLAINTEXT
  cloud:
    config:
      discovery:
@@ -21,6 +40,7 @@
    service-url:
      defaultZone: http://${EUREKA_URL:localhost}:8192/eureka


#实体加密、解密、字段脱敏拦截设置
domain:
  decrypt: true
@@ -40,4 +60,11 @@

  metrics:
    tags:
      application: property
      application: property

# kafka自定义消息发送配置
kafka:
  topic:
    kafka-demo-topic: Task_Info
    kafka-demo-key: order-key
    autoCreate: false