修改nettyTCP客户端断开重连 增加数据库连接池 修改日志打印

‘liusuyi’

2023-07-05 1fd64b07ddb99c2d9cc8a358b71aceb6a2c81492

修改nettyTCP客户端断开重连
增加数据库连接池
修改日志打印

 pom.xml

@@ -47,10 +47,7 @@
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.postgresql</groupId>
            <artifactId>postgresql</artifactId>
        </dependency>

        <!--海康录像机二次开发依赖jar包-->
        <dependency>
            <groupId>net.java.jna</groupId>
@@ -84,6 +81,20 @@
            <artifactId>geodesy</artifactId>
            <version>1.1.3</version>
        </dependency>
        <!--druid依赖-->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>druid-spring-boot-starter</artifactId>
            <version>1.2.14</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-jdbc</artifactId>
        </dependency>
        <dependency>
            <groupId>org.postgresql</groupId>
            <artifactId>postgresql</artifactId>
        </dependency>
    </dependencies>

    <build>

 src/main/java/com/ard/alarm/camera/service/impl/ArdCamerasServiceImpl.java

@@ -4,7 +4,7 @@
import com.ard.alarm.camera.domain.ArdCameras;
import com.ard.alarm.camera.mapper.ArdCamerasMapper;
import com.ard.alarm.camera.service.IArdCamerasService;
import com.ard.hiksdk.service.impl.HikClientUtil;
import com.ard.utils.hiksdk.service.impl.HikClientUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;


 src/main/java/com/ard/alarm/radar/service/impl/ArdEquipRadarServiceImpl.java

@@ -4,6 +4,7 @@
import com.ard.alarm.radar.domain.ArdEquipRadar;
import com.ard.alarm.radar.mapper.ArdEquipRadarMapper;
import com.ard.alarm.radar.service.IArdEquipRadarService;
import com.ard.utils.tcp.ClientInitialize;
import com.ard.utils.tcp.NettyTcpClient;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
@@ -19,32 +20,10 @@
 * @date 2023-03-11
 */
@Service
@Slf4j(topic = "radar")
public class ArdEquipRadarServiceImpl implements IArdEquipRadarService
{
    @Autowired
    private ArdEquipRadarMapper ardEquipRadarMapper;


    @Resource
    NettyTcpClient nettyTcpClient;
    @Value("${spring.netty.tcp.enabled}")
    private Boolean tcpClientEnable;
    @PostConstruct
    public void init() {
        if(!tcpClientEnable)
        {
            return;
        }
        List<ArdEquipRadar> ardEquipRadars = selectArdEquipRadarList(new ArdEquipRadar());
        for (ArdEquipRadar ardEquipRadar:ardEquipRadars)
        {
            String host = ardEquipRadar.getIp();
            Integer port = Integer.valueOf(ardEquipRadar.getPort());
            log.info("TCP客户端尝试连接："+host+":"+port);
            nettyTcpClient.init(ardEquipRadar);
        }
    }

    /**
     * 查询radar

 src/main/java/com/ard/alarm/tube/service/TubeAlarmService.java

@@ -39,7 +39,7 @@
            return;
        }
        nettyUdpServer.init(udpPort);
        log.info("UDP服务已启动");
        log.info("管线泄露UDP服务已启动");
    }

    @Async("alarm")

 src/main/java/com/ard/config/AsyncConfiguration.java

File was renamed from src/main/java/com/ard/async/AsyncConfiguration.java
@@ -1,4 +1,4 @@
package com.ard.async;
package com.ard.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@@ -16,7 +16,7 @@
@EnableAsync(proxyTargetClass = true)
public class AsyncConfiguration {
    @Bean("alarm")
    public ThreadPoolTaskExecutor executor(){
    public ThreadPoolTaskExecutor alarmExecutor(){
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        //配置核心线程数
        executor.setCorePoolSize(15);

 src/main/java/com/ard/config/DruidConfig.java

New file
@@ -0,0 +1,49 @@
package com.ard.config;
import com.alibaba.druid.filter.Filter;
import com.alibaba.druid.filter.stat.StatFilter;
import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.support.http.StatViewServlet;
import com.google.common.collect.Lists;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.web.servlet.ServletRegistrationBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
 * @Description: druid连接池的配置,配置如何处理慢sql，
 * @ClassName: DruidConfig
 * @Author: 刘苏义
 * @Date: 2023年07月04日15:29
 * @Version: 1.0
 **/
@Configuration
public class DruidConfig {
    //这个注解读取配置文件前缀为prefix的配置，将外部的配置文件与这里绑定
    @ConfigurationProperties(prefix = "spring.datasource.druid")
    //容器的开启与关闭
    @Bean(initMethod = "init",destroyMethod = "close")
    public DruidDataSource dataSource(){
        DruidDataSource dataSource=new DruidDataSource();
        dataSource.setProxyFilters(Lists.newArrayList(statFilter()));
        return  dataSource;
    }
    //bean注解，成为spring的bean，利用filter将慢sql的日志打印出来
    @Bean
    public Filter statFilter(){
        StatFilter statFilter=new StatFilter();
        //多长时间定义为慢sql，这里定义为5s
        statFilter.setSlowSqlMillis(5000);
        //是否打印出慢日志
        statFilter.setLogSlowSql(true);
        //是否将日志合并起来
        statFilter.setMergeSql(true);
        return  statFilter;
    }
    //这是配置druid的监控
    //这是配置druid的监控与配置文件中功能一样
    //#stat-view-servlet:
    //   #enabled:true
//    @Bean
//    public ServletRegistrationBean servletRegistrationBean(){
//        return new ServletRegistrationBean(new StatViewServlet(),"/druid/*");
//    }
}

 src/main/java/com/ard/config/MinioClientSingleton.java

@@ -1,12 +1,16 @@
package com.ard.config;

import com.ard.utils.SpringTool;
import io.minio.MinioClient;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.config.YamlPropertiesFactoryBean;
import org.springframework.context.annotation.Bean;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.Resource;
import org.springframework.integration.channel.DefaultHeaderChannelRegistry;
import org.springframework.integration.channel.PublishSubscribeChannel;
import org.springframework.stereotype.Component;

import java.io.IOException;
@@ -31,9 +35,9 @@
    private volatile static MinioClient minioClient;

    static {
        domainUrl = getYmlNew("minio.endpoint");
        accessKey = getYmlNew("minio.accessKey");
        secretKey = getYmlNew("minio.secretKey");
        domainUrl = SpringTool.getYmlInfo("minio.endpoint");
        accessKey = SpringTool.getYmlInfo("minio.accessKey");
        secretKey = SpringTool.getYmlInfo("minio.secretKey");
        log.info("minio信息：" + domainUrl + "(" + accessKey + "/" + secretKey + ")");
    }

@@ -56,20 +60,14 @@
        return minioClient;
    }

    /*yml配置信息获取*/
    public static String getYmlNew(String key) {
        Resource resource = new ClassPathResource("application.yml");
        Properties properties = null;
        try {
            YamlPropertiesFactoryBean yamlFactory = new YamlPropertiesFactoryBean();
            yamlFactory.setResources(resource);
            properties = yamlFactory.getObject();
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
        return properties.get(key).toString();
    }

    @Bean
    public PublishSubscribeChannel errorChannel() {
        return new PublishSubscribeChannel();
    }
    @Bean
    public DefaultHeaderChannelRegistry integrationHeaderChannelRegistry() {
        return new DefaultHeaderChannelRegistry();
    }
}


 src/main/java/com/ard/utils/ByteUtils.java

@@ -1,10 +1,9 @@
package com.ard.utils;

import javax.xml.bind.DatatypeConverter;
import java.io.ByteArrayOutputStream;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.zip.CRC32;

/**
 * @Description: 字节工具类
@@ -16,20 +15,20 @@
public class ByteUtils {
    /**
     * Byte字节转Hex
     *
     * @param b 字节
     * @return Hex
     */
    public static String byteToHex(byte b)
    {
    public static String byteToHex(byte b) {
        String hexString = Integer.toHexString(b & 0xFF);
        //由于十六进制是由0~9、A~F来表示1~16，所以如果Byte转换成Hex后如果是<16,就会是一个字符（比如A=10），通常是使用两个字符来表示16进制位的,
        //假如一个字符的话，遇到字符串11，这到底是1个字节，还是1和1两个字节，容易混淆，如果是补0，那么1和1补充后就是0101，11就表示纯粹的11
        if (hexString.length() < 2)
        {
        if (hexString.length() < 2) {
            hexString = new StringBuilder(String.valueOf(0)).append(hexString).toString();
        }
        return hexString.toUpperCase();
    }

    /**
     * byte数组转float
     */
@@ -37,6 +36,7 @@
        ByteBuffer buffer = ByteBuffer.wrap(bytes);
        return buffer.getFloat();
    }

    /**
     * byte数组转整型
     */
@@ -79,4 +79,98 @@
        return littleEndianBytes;
    }

    /**
     * 小端转大端
     */
    public static byte[] toBigEndian(byte[] littleEndianBytes) {
        byte[] bigEndianBytes = new byte[littleEndianBytes.length];
        for (int i = 0; i < littleEndianBytes.length; i++) {
            bigEndianBytes[i] = littleEndianBytes[littleEndianBytes.length - 1 - i];
        }
        return bigEndianBytes;
    }

    /**
     * int转byte数组
     */
    public static byte[] decimalToBytes(int decimalValue) {
        ByteBuffer buffer = ByteBuffer.allocate(Integer.BYTES);
        buffer.putInt(decimalValue);
        return buffer.array();
    }

    /**
     * byte数组拼接
     */
    public static byte[] appendArrays(byte[]... arrays) {
        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
        try {
            for (byte[] array : arrays) {
                outputStream.write(array);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return outputStream.toByteArray();
    }

    /**
     * byte数组CRC32校验
     */
    public static byte[] parseCrc32(byte[] bytes) {
        CRC32 crc32 = new CRC32();
        crc32.update(bytes);
        bytes = ByteUtils.decimalToBytes((int) crc32.getValue());
        bytes = ByteUtils.toBigEndian(bytes);
        // System.out.print("校验:"+DatatypeConverter.printHexBinary(bytes));//打印crc32的校验值
        return bytes;
    }

    /**
     * 去除包头包尾
     * 刘苏义
     * 2023/7/4 11:24
     */
    public static byte[] removeHeaderAndFooter(byte[] data, int headerLength, int footerLength) {
        int payloadLength = data.length - headerLength - footerLength;
        byte[] payload = new byte[payloadLength];
        System.arraycopy(data, headerLength, payload, 0, payloadLength);
        return payload;
    }

    /**
     * 去除包头包尾crc校验
     * 刘苏义
     * 2023/7/4 11:24
     */
    public static byte[] removeHeaderFooterAndCRC(byte[] data, int headerLength, int footerLength, int crcLength) {
        int payloadLength = data.length - headerLength - footerLength - crcLength;
        byte[] payload = new byte[payloadLength];
        System.arraycopy(data, headerLength, payload, 0, payloadLength);
        return payload;
    }

    /**
     * 获取后面的字节数组
     * 刘苏义
     * 2023/7/4 11:24
     */
    public static byte[] getLastBytes(byte[] data, int count) {
        int startIndex = data.length - count;
        byte[] lastBytes = new byte[count];
        System.arraycopy(data, startIndex, lastBytes, 0, count);
        return lastBytes;
    }

    /**
     * 找到00的索引位置
     */
    public static int findIndexOfDoubleZero(byte[] bytes) {
        for (int i = 0; i < bytes.length - 1; i++) {
            if (bytes[i] == 0x00) {
                return i;
            }
        }
        return -1;
    }
}

 src/main/java/com/ard/utils/SpringTool.java

@@ -1,11 +1,15 @@
package com.ard.utils;

import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.YamlPropertiesFactoryBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.Resource;
import org.springframework.stereotype.Component;

import java.nio.ByteBuffer;
import java.util.Properties;

/**
 * @Description:
@@ -36,4 +40,18 @@
        return getApplicationContext().getBean(name);
    }

    /*yml配置信息获取*/
    public static String getYmlInfo(String key) {
        Resource resource = new ClassPathResource("application.yml");
        Properties properties = null;
        try {
            YamlPropertiesFactoryBean yamlFactory = new YamlPropertiesFactoryBean();
            yamlFactory.setResources(resource);
            properties = yamlFactory.getObject();
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
        return properties.get(key).toString();
    }
}

 src/main/java/com/ard/utils/hiksdk/common/GlobalVariable.java

File was renamed from src/main/java/com/ard/hiksdk/common/GlobalVariable.java
@@ -1,4 +1,4 @@
package com.ard.hiksdk.common;
package com.ard.utils.hiksdk.common;

import com.ard.alarm.camera.domain.ArdCameras;
import lombok.Data;

 src/main/java/com/ard/utils/hiksdk/domain/alarmEventInfo.java

File was renamed from src/main/java/com/ard/hiksdk/domain/alarmEventInfo.java
@@ -1,4 +1,4 @@
package com.ard.hiksdk.domain;
package com.ard.utils.hiksdk.domain;

import lombok.Data;


 src/main/java/com/ard/utils/hiksdk/domain/recordInfo.java

File was renamed from src/main/java/com/ard/hiksdk/domain/recordInfo.java
@@ -1,4 +1,4 @@
package com.ard.hiksdk.domain;
package com.ard.utils.hiksdk.domain;

import lombok.AllArgsConstructor;
import lombok.Data;

 src/main/java/com/ard/utils/hiksdk/service/impl/FMSGCallBack.java

File was renamed from src/main/java/com/ard/hiksdk/service/impl/FMSGCallBack.java
@@ -1,14 +1,14 @@
package com.ard.hiksdk.service.impl;
package com.ard.utils.hiksdk.service.impl;

import com.alibaba.fastjson2.JSON;
import com.ard.alarm.camera.domain.ArdCameras;
import com.ard.alarm.camera.domain.CameraCmd;
import com.ard.alarm.camera.service.impl.ArdCamerasServiceImpl;
import com.ard.config.MinioClientSingleton;
import com.ard.hiksdk.common.GlobalVariable;
import com.ard.hiksdk.domain.alarmEventInfo;
import com.ard.hiksdk.util.hikSdkUtil.HCNetSDK;
import com.ard.hiksdk.util.minio.MinioUtils;
import com.ard.utils.hiksdk.common.GlobalVariable;
import com.ard.utils.hiksdk.domain.alarmEventInfo;
import com.ard.utils.hiksdk.util.hikSdkUtil.HCNetSDK;
import com.ard.utils.hiksdk.util.minio.MinioUtils;
import com.ard.utils.DateUtils;
import com.ard.utils.SpringTool;
import com.ard.utils.mqtt.MqttConsumer;
@@ -133,7 +133,10 @@
                publishMqtt(info);
                break;
            case HCNetSDK.COMM_UPLOAD_FACESNAP_RESULT:
                log.info("人脸检测事件");
                log.info("人脸检测事件上报");
                break;
            case HCNetSDK.COMM_GISINFO_UPLOAD:
                log.info("GPS报警信息上报");
                break;
            default:
                log.info("未知报警事件类型:lCommand:" + Integer.toHexString(lCommand));

 src/main/java/com/ard/utils/hiksdk/service/impl/HikClientUtil.java

File was renamed from src/main/java/com/ard/hiksdk/service/impl/HikClientUtil.java
@@ -1,10 +1,10 @@
package com.ard.hiksdk.service.impl;
package com.ard.utils.hiksdk.service.impl;

import com.ard.alarm.camera.domain.ArdCameras;
import com.ard.alarm.camera.domain.CameraCmd;
import com.ard.hiksdk.common.GlobalVariable;
import com.ard.hiksdk.util.hikSdkUtil.HCNetSDK;
import com.ard.hiksdk.util.minio.MinioUtils;
import com.ard.utils.hiksdk.common.GlobalVariable;
import com.ard.utils.hiksdk.util.hikSdkUtil.HCNetSDK;
import com.ard.utils.hiksdk.util.minio.MinioUtils;
import com.sun.jna.Native;
import com.sun.jna.Platform;
import com.sun.jna.ptr.IntByReference;
@@ -14,7 +14,7 @@
import java.nio.ByteBuffer;
import java.util.*;

import static com.ard.hiksdk.common.GlobalVariable.cameraMap;
import static com.ard.utils.hiksdk.common.GlobalVariable.cameraMap;


/**

 src/main/java/com/ard/utils/hiksdk/service/impl/LoginResultCallBack.java

File was renamed from src/main/java/com/ard/hiksdk/service/impl/LoginResultCallBack.java
@@ -1,8 +1,8 @@
package com.ard.hiksdk.service.impl;
package com.ard.utils.hiksdk.service.impl;

import com.ard.alarm.camera.domain.ArdCameras;
import com.ard.hiksdk.common.GlobalVariable;
import com.ard.hiksdk.util.hikSdkUtil.HCNetSDK;
import com.ard.utils.hiksdk.common.GlobalVariable;
import com.ard.utils.hiksdk.util.hikSdkUtil.HCNetSDK;
import com.sun.jna.Pointer;
import lombok.extern.slf4j.Slf4j;


 src/main/java/com/ard/utils/hiksdk/util/hikSdkUtil/HCNetSDK.java

File was renamed from src/main/java/com/ard/hiksdk/util/hikSdkUtil/HCNetSDK.java
@@ -1,4 +1,4 @@
package com.ard.hiksdk.util.hikSdkUtil;
package com.ard.utils.hiksdk.util.hikSdkUtil;

import com.sun.jna.*;
import com.sun.jna.examples.win32.W32API;
@@ -899,6 +899,7 @@
    public static final int COMM_ALARM_V30 = 0x4000;//9000报警信息主动上传
    public static final int COMM_ALARM_V40 = 0x4007;
    public static final int COMM_ALARM_RULE = 0x1102;//行为分析信息上传
    public static final int COMM_GISINFO_UPLOAD = 0x4012;//GIS信息上传
    public static final int COMM_ALARM_PDC = 0x1103;//客流量统计报警上传
    public static final int COMM_UPLOAD_PLATE_RESULT = 0x2800;//交通抓拍结果上传
    public static final int COMM_ITS_PLATE_RESULT = 0x3050;//交通抓拍的终端图片上传

 src/main/java/com/ard/utils/hiksdk/util/hikSdkUtil/HIKSDKStructure.java

File was renamed from src/main/java/com/ard/hiksdk/util/hikSdkUtil/HIKSDKStructure.java
@@ -1,4 +1,4 @@
package com.ard.hiksdk.util.hikSdkUtil;
package com.ard.utils.hiksdk.util.hikSdkUtil;

import com.sun.jna.Structure;


 src/main/java/com/ard/utils/hiksdk/util/imageUtil/waterMarkUtil.java

File was renamed from src/main/java/com/ard/hiksdk/util/imageUtil/waterMarkUtil.java
@@ -1,4 +1,4 @@
package com.ard.hiksdk.util.imageUtil;
package com.ard.utils.hiksdk.util.imageUtil;

/**
 * @ClassName WaterMarkUtil

 src/main/java/com/ard/utils/hiksdk/util/minio/MinioUtils.java

File was renamed from src/main/java/com/ard/hiksdk/util/minio/MinioUtils.java
@@ -1,4 +1,4 @@
package com.ard.hiksdk.util.minio;
package com.ard.utils.hiksdk.util.minio;

import com.ard.config.MinioClientSingleton;
import io.minio.*;

 src/main/java/com/ard/utils/mqtt/MqttConsumer.java

@@ -165,9 +165,9 @@
                log.info("消息发送成功");
            }
        } catch (MqttPersistenceException e) {
            e.printStackTrace();
            log.error("mqtt持久异常：" + e.getMessage());
        } catch (MqttException e) {
            e.printStackTrace();
            log.error("mqtt异常：" + e.getMessage());
        }
    }
}

 src/main/java/com/ard/utils/tcp/ClientHandler.java

New file
@@ -0,0 +1,444 @@
package com.ard.utils.tcp;

/**
 * @Description:
 * @ClassName: ClientHandler
 * @Author: 刘苏义
 * @Date: 2023年07月05日13:13
 * @Version: 1.0
 **/

import com.alibaba.fastjson2.JSON;
import com.ard.alarm.radar.domain.ArdAlarmRadar;
import com.ard.alarm.radar.domain.ArdEquipRadar;
import com.ard.alarm.radar.domain.RadarAlarmData;
import com.ard.utils.ByteUtils;
import com.ard.utils.GisUtils;
import com.ard.utils.mqtt.MqttConsumer;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelId;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import javax.xml.bind.DatatypeConverter;
import java.net.InetSocketAddress;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

/**
 * 客户端处理器
 *
 * @author lijiamin
 */
@Slf4j(topic = "netty")
public class ClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
    private ChannelHandlerContext context;
    private ScheduledFuture<?> heartbeatTask;

    /**
     * 连接建立
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        context = ctx;
        startHeartbeatTask();//开始发送心跳
    }

    /**
     * 连接断开
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        ChannelId id = ctx.channel().id();
        InetSocketAddress ipSocket = (InetSocketAddress) ctx.channel().remoteAddress();
        int port = ipSocket.getPort();
        String host = ipSocket.getHostString();
        log.error("与设备" + host + ":" + port + "连接断开!");
        ArdEquipRadar ardEquipRadar = ClientInitialize.tureConnectMap.get(id);
        // 连接断开后的最后处理
        ctx.pipeline().remove(this);
        ctx.deregister();
        ctx.close();

        // 将失败信息插入Set集合
        ClientInitialize.falseConnectSet.add(ardEquipRadar);
        super.channelInactive(ctx);
    }

    /**
     * 通道数据读取
     * 业务操作
     *
     * @param ctx
     * @param msg
     * @throws Exception
     */
    @Override
    public void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
        ChannelId id = ctx.channel().id();
        ArdEquipRadar ardEquipRadar = ClientInitialize.tureConnectMap.get(id);
        // 处理接收到的消息
        byte[] byteArray = new byte[msg.readableBytes()];
        msg.getBytes(msg.readerIndex(), byteArray);
        byte[] bytes = MessageParsing.receiveCompletePacket(byteArray);
        if (bytes != null) {
            processData(ardEquipRadar, bytes);
        }
    }

    /**
     * 通道数据处理完成
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        super.channelReadComplete(ctx);
    }

    /**
     * 事件触发
     *
     * @param ctx
     * @param evt
     * @throws Exception
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        super.userEventTriggered(ctx, evt);
    }

    /**
     * 异常触发
     *
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        // 发生异常时的处理
        cause.printStackTrace();
        ctx.close();
        stopHeartbeatTask();//停止心跳发送
    }

    /**
     * 开始心跳任务
     */
    private void startHeartbeatTask() {
        heartbeatTask = context.executor().scheduleAtFixedRate(() -> {
            // 发送心跳消息
            ByteBuf message = context.alloc().buffer();
            byte[] header = {0x01, 0x02, 0x01};
            byte[] payload = {0x10, 0x00, 0x00, 0x00};
            byte[] payloadCrc32 = ByteUtils.parseCrc32(payload);
            byte[] footer = {0x01, 0x02, 0x00};
            byte[] heart = ByteUtils.appendArrays(header, payload, payloadCrc32, footer);
//            byte[] heart = {0x01, 0x02, 0x01, 0x10, 0x00, 0x00, 0x00, (byte) 0x83, (byte) 0x88, 0x5d, 0x71, 0x01, 0x02, 0x00};
            String hexString = DatatypeConverter.printHexBinary(heart);
            log.info("发送心跳:" + hexString);
            message.writeBytes(heart);
            context.writeAndFlush(message);

        }, 0, 5, TimeUnit.SECONDS);
    }

    /**
     * 停止心跳任务
     */
    private void stopHeartbeatTask() {
        if (heartbeatTask != null) {
            heartbeatTask.cancel(false);
            heartbeatTask = null;
        }
    }

    /**
     * 解析报警数据
     */
    public void processData(ArdEquipRadar ardEquipRadarbyte, byte[] data) {
        try {
            String radarId = ardEquipRadarbyte.getId();
            String radarName = ardEquipRadarbyte.getName();
            Double radarLongitude = ardEquipRadarbyte.getLongitude();
            Double radarLagitude = ardEquipRadarbyte.getLatitude();
            //region crc校验-目前仅用于显示校验结果
            Boolean crc32Check = MessageParsing.CRC32Check(data);
            if (!crc32Check) {
                log.info("CRC32校验不通过");
            } else {
                log.info("CRC32校验通过");
            }
            //endregion
            //log.info("原始数据:" + DatatypeConverter.printHexBinary(data));
            //log.info("雷达信息：" + host + "【port】" + port + "【X】" + longitude + "【Y】" + lagitude + "【Z】" + altitude);
            data = MessageParsing.transferData(data);//去掉包头和包尾、校验及转义
            //region 负载头解析
            byte[] type = Arrays.copyOfRange(data, 0, 1);//命令类型
            //  log.info("命令类型:" + DatatypeConverter.printHexBinary(type));
            byte[] cmdId = Arrays.copyOfRange(data, 1, 2);//命令ID
            //  log.info("命令ID:" + DatatypeConverter.printHexBinary(cmdId));
            byte[] payloadSize = Arrays.copyOfRange(data, 2, 4);//有效负载大小
            payloadSize = ByteUtils.toLittleEndian(payloadSize);
            int payloadSizeToDecimal = ByteUtils.bytesToDecimal(payloadSize);
            // log.info("有效负载大小(转整型):" + payloadSizeToDecimal);
            //endregion
            List<ArdAlarmRadar> radarAlarmInfos = new ArrayList<>();
            String alarmTime = "";
            Integer targetNum = 0;
            if (Arrays.equals(cmdId, new byte[]{0x01})) {
                //region 告警信息反馈
                byte[] dwTim = Arrays.copyOfRange(data, 4, 8);
                dwTim = ByteUtils.toLittleEndian(dwTim);
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                long l = ByteUtils.bytesToDecimal(dwTim);
                alarmTime = sdf.format(l * 1000);
                // log.info("周视图像的出现时间(转date):" + alarmTime);

                byte[] wTargetNum = Arrays.copyOfRange(data, 8, 10);
                wTargetNum = ByteUtils.toLittleEndian(wTargetNum);
                targetNum = ByteUtils.bytesToDecimal(wTargetNum);
                if (targetNum == 0) {
                    return;
                }
                log.info("目标总点数(转整型):" + targetNum);

                //解析NET_TARGET_UNIT(64是NET_TARGET_HEAD的字节数)
                int uintSize = (payloadSizeToDecimal - 64) / targetNum;
                // log.info("单条报警大小:" + uintSize);

                for (int i = 0; i < targetNum; i++) {

                    Integer index = 68 + uintSize * i;
                    byte[] dwID = Arrays.copyOfRange(data, index, index + 4);
                    // log.info("目标ID:" + DatatypeConverter.printHexBinary(cmdId));
                    dwID = ByteUtils.toLittleEndian(dwID);
                    int id = ByteUtils.bytesToDecimal(dwID);

                    // log.info("目标ID号：" + id);

                    byte[] iDistance = Arrays.copyOfRange(data, index + 8, index + 12);
                    iDistance = ByteUtils.toLittleEndian(iDistance);
                    Double Distance = ByteUtils.bytesToDouble(iDistance);
                    // log.info("目标当前距离(m):" + Distance);
                    //region 不需要的字段
//                    byte[] dwGSum = Arrays.copyOfRange(data, index + 4, index + 8);
//                    dwGSum = toLittleEndian(dwGSum);
//                    int GSum = byteArrayToDecimal(dwGSum);
//                    log.info("目标当前像素灰度和：" + GSum);
//                    byte[] iTw = Arrays.copyOfRange(data, index + 12, index + 16);
//                    iTw = toLittleEndian(iTw);
//                    int Tw = byteArrayToDecimal(iTw);
//                    log.info("目标当前的像素宽度:" + Tw);
//
//                    byte[] iTh = Arrays.copyOfRange(data, index + 16, index + 20);
//                    iTh = toLittleEndian(iTh);
//                    int Th = byteArrayToDecimal(iTh);
//                    log.info("目标当前的像素高度:" + Th);
//
//                    byte[] wPxlArea = Arrays.copyOfRange(data, index + 20, index + 22);
//                    wPxlArea = toLittleEndian(wPxlArea);
//                    int PxlArea = byteArrayToDecimal(wPxlArea);
//                    log.info("目标当前像素面积:" + PxlArea);
//
//                    byte[] cTrkNum = Arrays.copyOfRange(data, index + 22, index + 23);
//                    cTrkNum = toLittleEndian(cTrkNum);
//                    int TrkNum = byteArrayToDecimal(cTrkNum);
//                    log.info("轨迹点数:" + TrkNum);

//                    byte[] sVx = Arrays.copyOfRange(data, index + 24, index + 26);
//                    sVx = toLittleEndian(sVx);
//                    int Vx = byteArrayToDecimal(sVx);
//                    log.info("目标当前速度矢量(像素距离)X:" + Vx);
//
//                    byte[] sVy = Arrays.copyOfRange(data, index + 26, index + 28);
//                    sVy = toLittleEndian(sVy);
//                    int Vy = byteArrayToDecimal(sVy);
//                    log.info("目标当前速度矢量(像素距离)Y:" + Vy);
//
//                    byte[] sAreaNo = Arrays.copyOfRange(data, index + 28, index + 30);
//                    sAreaNo = toLittleEndian(sAreaNo);
//                    int AreaNo = byteArrayToDecimal(sAreaNo);
//                    log.info("目标归属的告警区域号:" + AreaNo);
//
//                    byte[] cGrp = Arrays.copyOfRange(data, index + 30, index + 31);
//                    cGrp = toLittleEndian(cGrp);
//                    int Grp = byteArrayToDecimal(cGrp);
//                    log.info("所属组:" + Grp);
                    //endregion
                    String alarmType = "";
                    byte[] cStat = Arrays.copyOfRange(data, index + 23, index + 24);
                    cStat = ByteUtils.toLittleEndian(cStat);
                    String binaryString = String.format("%8s", Integer.toBinaryString(cStat[0] & 0xFF)).replace(' ', '0');
                    // log.info("目标当前状态:" + binaryString);
                    // 提取第4位至第6位的值
                    int extractedValue = (cStat[0] >> 4) & 0b00001111;
                    // 判断提取的值
                    if (extractedValue == 0b0000) {
                        alarmType = "运动目标检测";
                    } else if (extractedValue == 0b0001) {
                        alarmType = "热源检测";
                    }
                    // log.info("报警类型:" + alarmType);

                    byte[] szName = Arrays.copyOfRange(data, index + 64, index + 96);
                    int position = ByteUtils.findIndexOfDoubleZero(szName);
                    String alarmPointName;
                    if (position != -1) {
                        byte[] result = new byte[position];
                        System.arraycopy(szName, 0, result, 0, position);
                        alarmPointName = new String(result, "GBK");
                    } else {
                        alarmPointName = new String(szName, "GBK");
                    }
                    // log.info("所属告警区域名称:" + alarmPointName);
                    byte[] afTx = Arrays.copyOfRange(data, index + 96, index + 100);
                    afTx = ByteUtils.toLittleEndian(afTx);
                    float fTx = ByteUtils.bytesToFloat(afTx);
                    //  log.info("水平角度:" + fTx);
                    byte[] afTy = Arrays.copyOfRange(data, index + 112, index + 116);
                    afTy = ByteUtils.toLittleEndian(afTy);
                    float fTy = ByteUtils.bytesToFloat(afTy);
                    //  log.info("垂直角度:" + fTy);
                    Double[] radarXY = {radarLagitude, radarLagitude};
                    Double[] alarmXY = GisUtils.CalculateCoordinates(radarXY, Distance, (double) fTx);
                    //  log.info("报警信息：" + "【id】" + id + "【name】" + alarmPointName + "【alarmType】" + alarmType + "【alarmTime】" + alarmTime + "【distance】" + Distance + "【P】" + fTx + "【T】" + fTy + "【X】" + alarmXY[0] + "【Y】" + alarmXY[1]);
                    ArdAlarmRadar ardAlarmRadar = new ArdAlarmRadar();
                    ardAlarmRadar.setTargetId(id);
                    ardAlarmRadar.setName(alarmPointName);
                    ardAlarmRadar.setLongitude(alarmXY[0]);
                    ardAlarmRadar.setLatitude(alarmXY[1]);
                    ardAlarmRadar.setAlarmType(alarmType);
                    radarAlarmInfos.add(ardAlarmRadar);
                }
                //endregion
            }
            if (Arrays.equals(cmdId, new byte[]{0x04})) {
                //region抽油机AI状态反馈
                String hexString = DatatypeConverter.printHexBinary(data);
                //log.info(hexString);

                byte[] dwTim = Arrays.copyOfRange(data, 4, 8);
                dwTim = ByteUtils.toLittleEndian(dwTim);
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                long l = ByteUtils.bytesToDecimal(dwTim);
                alarmTime = sdf.format(l * 1000);
                //log.info("周视图像的出现时间(转date):" + alarmTime);

                byte[] wTargetNum = Arrays.copyOfRange(data, 8, 10);
                wTargetNum = ByteUtils.toLittleEndian(wTargetNum);
                targetNum = ByteUtils.bytesToDecimal(wTargetNum);
                log.info("目标总点数(转整型):" + targetNum);
                if (targetNum == 0) {
                    return;
                }
                //解析NET_TARGET_UNIT(64是NET_TARGET_HEAD的字节数)
                int uintSize = (payloadSizeToDecimal - 64) / targetNum;
                //log.info("单条报警大小:" + uintSize);
                for (int i = 0; i < targetNum; i++) {
                    Integer index = 68 + uintSize * i;
                    byte[] dwID = Arrays.copyOfRange(data, index, index + 4);
                    //log.info("目标ID:" + DatatypeConverter.printHexBinary(dwID));
                    dwID = ByteUtils.toLittleEndian(dwID);
                    int id = ByteUtils.bytesToDecimal(dwID);
                    //log.info("目标ID号：" + id);
                    //region 不需要的字段
                    byte[] iTw = Arrays.copyOfRange(data, index + 4, index + 8);
                    iTw = ByteUtils.toLittleEndian(iTw);
                    int Tw = ByteUtils.bytesToDecimal(iTw);
                    // log.info("目标当前的像素宽度:" + Tw);

                    byte[] iTh = Arrays.copyOfRange(data, index + 8, index + 12);
                    iTh = ByteUtils.toLittleEndian(iTh);
                    int Th = ByteUtils.bytesToDecimal(iTh);
                    //log.info("目标当前的像素高度:" + Th);

                    byte[] fTx = Arrays.copyOfRange(data, index + 12, index + 16);
                    fTx = ByteUtils.toLittleEndian(fTx);
                    float fTxAngle = ByteUtils.bytesToFloat(fTx);
                    //log.info("p角度:" + fTxAngle);
                    byte[] fTy = Arrays.copyOfRange(data, index + 16, index + 20);
                    fTy = ByteUtils.toLittleEndian(fTy);
                    float fTyAngle = ByteUtils.bytesToFloat(fTy);
                    //log.info("t角度:" + fTyAngle);

                    byte[] sAreaNo = Arrays.copyOfRange(data, index + 20, index + 22);
                    sAreaNo = ByteUtils.toLittleEndian(sAreaNo);
                    int AreaNo = ByteUtils.bytesToDecimal(sAreaNo);
                    //log.info("目标归属的告警区域号:" + AreaNo);

                    byte[] cGrp = Arrays.copyOfRange(data, index + 22, index + 23);
                    cGrp = ByteUtils.toLittleEndian(cGrp);
                    int Grp = ByteUtils.bytesToDecimal(cGrp);
                    //log.info("所属组:" + Grp);
                    //endregion
                    String alarmType;
                    byte[] cStat = Arrays.copyOfRange(data, index + 23, index + 24);
                    cStat = ByteUtils.toLittleEndian(cStat);
                    //String binaryString = String.format("%8s", Integer.toBinaryString(cStat[0] & 0xFF)).replace(' ', '0');
                    //log.info("目标当前状态:" + binaryString);
                    // 提取第0位值
                    // 使用位运算操作判断第0位是否为1
                    boolean isB0 = (cStat[0] & 0x01) == 0x00;
                    // 判断提取的值
                    if (isB0) {
                        alarmType = "雷达抽油机停机";
                    } else {
                        continue;
                    }
                    //log.info("报警类型:" + alarmType);

                    byte[] szName = Arrays.copyOfRange(data, index + 32, index + 64);
                    //log.info("所属告警区域名称:" + DatatypeConverter.printHexBinary(szName));
                    int position = ByteUtils.findIndexOfDoubleZero(szName);
                    String alarmPointName;
                    if (position != -1) {
                        byte[] result = new byte[position];
                        System.arraycopy(szName, 0, result, 0, position);
                        alarmPointName = new String(result, "GBK");
                    } else {
                        alarmPointName = new String(szName, "GBK");
                    }
                    // log.info("所属告警区域名称:" + alarmPointName);
                    //log.info("报警信息：" + "【id】" + id + "【name】" + alarmPointName + "【alarmType】" + alarmType + "【alarmTime】" + alarmTime);
                    ArdAlarmRadar ardAlarmRadar = new ArdAlarmRadar();
                    ardAlarmRadar.setTargetId(id);
                    ardAlarmRadar.setName(alarmPointName);
                    ardAlarmRadar.setAlarmType(alarmType);
                    radarAlarmInfos.add(ardAlarmRadar);
                }
                //endregion
            }
            if (StringUtils.isEmpty(alarmTime)) {
                return;
            }
            if (targetNum == 0) {
                return;
            }
            RadarAlarmData radarAlarmData = new RadarAlarmData();
            radarAlarmData.setRadarId(radarId);
            radarAlarmData.setRadarName(radarName);
            radarAlarmData.setAlarmTime(alarmTime);
            radarAlarmData.setArdAlarmRadars(radarAlarmInfos);
            MqttConsumer.publish(2, false, "radar", JSON.toJSONString(radarAlarmData));
        } catch (Exception ex) {
            log.error("雷达报文解析异常:" + ex.getMessage());
        }
    }

}

 src/main/java/com/ard/utils/tcp/ClientInitialize.java

New file
@@ -0,0 +1,136 @@
package com.ard.utils.tcp;

/**
 * @Description:
 * @ClassName: init
 * @Author: 刘苏义
 * @Date: 2023年07月05日13:11
 * @Version: 1.0
 **/

import com.ard.alarm.radar.domain.ArdEquipRadar;
import com.ard.alarm.radar.service.IArdEquipRadarService;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.function.Consumer;

/**
 * 客户端初始化
 *
 * @author lijiamin
 */
@Component
@Slf4j(topic = "netty")
public class ClientInitialize {

    @Resource
    IArdEquipRadarService ardEquipRadarService;

    @Value("${spring.netty.tcp.enabled}")
    private Boolean tcpClientEnable;

    private Bootstrap bootstrap;
    public static CopyOnWriteArraySet<ArdEquipRadar> falseConnectSet = new CopyOnWriteArraySet();
    public static ConcurrentHashMap<ChannelId, ArdEquipRadar> tureConnectMap = new ConcurrentHashMap();

    /**
     * Netty初始化配置
     */
    public void initNettyTcp() {
        EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        bootstrap = new Bootstrap();
        bootstrap.group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) {
                        socketChannel.pipeline().addLast(new ClientHandler());
                    }
                });

        // 异步持续监听连接失败的地址
        CompletableFuture.runAsync(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    try {
                        if (falseConnectSet.size() != 0) {
                            // 循环集合内元素
                            falseConnectSet.forEach(new Consumer<ArdEquipRadar>() {
                                @Override
                                public void accept(ArdEquipRadar ardEquipRadar) {
                                    connectServer(ardEquipRadar);
                                }
                            });
                        }
                        Thread.sleep(3000);
                    } catch (Exception e) {
                        log.error("Netty初始化配置监听地址出现异常");
                        e.printStackTrace();
                    }
                }
            }
        });
    }

    /**
     * 服务连接
     *
     * @param ardEquipRadar
     */
    public void connectServer(ArdEquipRadar ardEquipRadar) {
        // 获取地址及端口
        String host = ardEquipRadar.getIp();
        Integer port = ardEquipRadar.getPort();
        // 异步连接tcp服务端
        bootstrap.remoteAddress(host, port).connect().addListener((ChannelFuture futureListener) -> {
            if (!futureListener.isSuccess()) {
                log.error(host + ":" + port + "雷达TCP服务端连接失败");
                futureListener.channel().close();
                // 连接失败信息插入Set
                falseConnectSet.add(ardEquipRadar);
                // 连接失败信息从map移除
                tureConnectMap.remove(futureListener.channel().id());
            } else {
                log.info(host + ":" + port + "雷达TCP服务端连接成功");
                // 连接成功信息从Set拔除
                falseConnectSet.remove(ardEquipRadar);
                // 连接成功信息写入map
                tureConnectMap.put(futureListener.channel().id(), ardEquipRadar);
            }
        });
    }

    /**
     * 初始化方法
     */
    @PostConstruct
    public void initialize() {
        if (!tcpClientEnable) {
            return;
        }
        initNettyTcp();//初始化nettyTcp
        List<ArdEquipRadar> ardEquipRadars = ardEquipRadarService.selectArdEquipRadarList(new ArdEquipRadar());
        for (ArdEquipRadar ardEquipRadar : ardEquipRadars) {
            String host = ardEquipRadar.getIp();
            Integer port = Integer.valueOf(ardEquipRadar.getPort());
            log.info("TCP客户端尝试连接：" + host + ":" + port);
            connectServer(ardEquipRadar);//连接每一个雷达服务
        }
    }
}

 src/main/java/com/ard/utils/tcp/MessageParsing.java

@@ -1,5 +1,7 @@
package com.ard.utils.tcp;

import com.ard.utils.ByteUtils;

import javax.xml.bind.DatatypeConverter;
import java.util.ArrayList;
import java.util.Arrays;
@@ -44,6 +46,36 @@
    }

    /**
     * crc32校验检查
     * 刘苏义
     * 2023/7/4 11:24
     */
    public static Boolean CRC32Check(byte[] packet) {
        //System.out.println(DatatypeConverter.printHexBinary(packet));
        int headerLength = 3;//包头3个字节
        int footerLength = 3;//包尾3个字节
        int crcLength = 4;//crc校验4个字节
        //去掉包头包尾
        byte[] payloadCrc32 = ByteUtils.removeHeaderAndFooter(packet, headerLength, footerLength);
        //System.out.println(DatatypeConverter.printHexBinary(payloadCrc32));
        //获取到数据携带的crc32值
        byte[] oldCrc32 = ByteUtils.getLastBytes(payloadCrc32, crcLength);
        //去掉包头包尾crc32字节，仅保留负载
        byte[] payload = ByteUtils.removeHeaderFooterAndCRC(packet, headerLength, footerLength, crcLength);
       // System.out.println(DatatypeConverter.printHexBinary(payload));
        //计算负载的crc32值
        byte[] NewCrc32 = ByteUtils.parseCrc32(payload);
        System.out.println("old:"+DatatypeConverter.printHexBinary(oldCrc32));
        System.out.println("new:"+DatatypeConverter.printHexBinary(NewCrc32));
        //判断数据的crc32校验值和计算值是否相同
        if (Arrays.equals(oldCrc32, NewCrc32)) {
            return true;
        } else {
            return false;
        }
    }

    /**
     * 获取包结束索引
     */
    public static int findPacketEndIndex(List<Byte> buffer, byte[] packetEnd) {
@@ -78,8 +110,10 @@
     * 去掉包头和包尾校验及转义
     */
    public static byte[] transferData(byte[] data) {
        data = Arrays.copyOfRange(data, 3, data.length);
        data = Arrays.copyOfRange(data, 0, data.length - 7);
        int headerLength = 3;//包头3个字节
        int footerLength = 3;//包尾3个字节
        int crcLength = 4;//crc校验4个字节
        data= ByteUtils.removeHeaderFooterAndCRC(data,headerLength,footerLength,crcLength);
        String dataStr = DatatypeConverter.printHexBinary(data);
        if (dataStr.contains("01020201")) {//转义01020201
            dataStr = dataStr.replaceAll("01020201", "010201");

 src/main/java/com/ard/utils/tcp/NettyTcpClient.java

@@ -16,7 +16,7 @@
import java.util.concurrent.TimeUnit;

/**
 * @Description:
 * @Description: 另一种tcp客户端 目前弃用
 * @ClassName: NettyTcpClient
 * @Author: 刘苏义
 * @Date: 2023年06月25日17:00
@@ -24,11 +24,10 @@
 **/
@EnableAsync
@Component
@Slf4j(topic = "radar")
@Slf4j(topic = "netty")
public class NettyTcpClient {
    @Async
    @Async("alarm")
    public void init(ArdEquipRadar ardEquipRadar) {

        while (true) {
            NioEventLoopGroup group = new NioEventLoopGroup();
            try {

 src/main/java/com/ard/utils/tcp/NettyTcpClientHandler.java

@@ -11,20 +11,22 @@
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import javax.xml.bind.DatatypeConverter;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;


/**
 * @Description: tcp客户端处理
 * @Description: tcp客户端处理-弃用
 * @ClassName: NettyTcpClientHandler
 * @Author: 刘苏义
 * @Date: 2023年06月25日17:02
 * @Version: 1.0
 **/
@Slf4j(topic = "radar")
@Slf4j(topic = "netty")
public class NettyTcpClientHandler extends SimpleChannelInboundHandler<ByteBuf> {

    private String host;
@@ -72,7 +74,12 @@
        heartbeatTask = context.executor().scheduleAtFixedRate(() -> {
            // 发送心跳消息
            ByteBuf message = context.alloc().buffer();
            byte[] heart = {0x01, 0x02, 0x01, 0x10, 0x00, 0x00, 0x00, (byte) 0x83, (byte) 0x88, 0x5d, 0x71, 0x01, 0x02, 0x00};
            byte[] header = {0x01, 0x02, 0x01};
            byte[] payload = {0x10, 0x00, 0x00, 0x00};
            byte[] payloadCrc32 = ByteUtils.parseCrc32(payload);
            byte[] footer = {0x01, 0x02, 0x00};
            byte[] heart = ByteUtils.appendArrays(header, payload, payloadCrc32, footer);
//            byte[] heart = {0x01, 0x02, 0x01, 0x10, 0x00, 0x00, 0x00, (byte) 0x83, (byte) 0x88, 0x5d, 0x71, 0x01, 0x02, 0x00};
            String hexString = DatatypeConverter.printHexBinary(heart);
            log.info("发送心跳:" + hexString);
            message.writeBytes(heart);
@@ -105,23 +112,32 @@
     */
    public void processData(byte[] data) {
        try {
            //region crc校验-目前仅用于显示校验结果
            Boolean crc32Check = MessageParsing.CRC32Check(data);
            if (!crc32Check) {
                log.info("CRC32校验不通过");
            } else {
                log.info("CRC32校验通过");
            }
            //endregion
            //log.info("原始数据:" + DatatypeConverter.printHexBinary(data));
            log.info("雷达信息：" + host + "【port】" + port + "【X】" + longitude + "【Y】" + lagitude + "【Z】" + altitude);
            //log.info("雷达信息：" + host + "【port】" + port + "【X】" + longitude + "【Y】" + lagitude + "【Z】" + altitude);
            data = MessageParsing.transferData(data);//去掉包头和包尾、校验及转义

            //region 负载头解析
            byte[] type = Arrays.copyOfRange(data, 0, 1);//命令类型
            //  log.info("命令类型:" + DatatypeConverter.printHexBinary(type));

            byte[] cmdId = Arrays.copyOfRange(data, 1, 2);//命令ID
            //  log.info("命令ID:" + DatatypeConverter.printHexBinary(cmdId));

            byte[] payloadSize = Arrays.copyOfRange(data, 2, 4);//有效负载大小
            payloadSize = ByteUtils.toLittleEndian(payloadSize);
            int payloadSizeToDecimal = ByteUtils.bytesToDecimal(payloadSize);
            // log.info("有效负载大小(转整型):" + payloadSizeToDecimal);
            //endregion
            List<ArdAlarmRadar> radarAlarmInfos = new ArrayList<>();
            String alarmTime="";
            String alarmTime = "";
            Integer targetNum = 0;
            if (Arrays.equals(cmdId, new byte[]{0x01})) {
                //region 告警信息反馈
                byte[] dwTim = Arrays.copyOfRange(data, 4, 8);
                dwTim = ByteUtils.toLittleEndian(dwTim);
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
@@ -131,8 +147,11 @@

                byte[] wTargetNum = Arrays.copyOfRange(data, 8, 10);
                wTargetNum = ByteUtils.toLittleEndian(wTargetNum);
                int targetNum = ByteUtils.bytesToDecimal(wTargetNum);
                //log.info("目标总点数(转整型):" + targetNum);
                targetNum = ByteUtils.bytesToDecimal(wTargetNum);
                if (targetNum == 0) {
                    return;
                }
                log.info("目标总点数(转整型):" + targetNum);

                //解析NET_TARGET_UNIT(64是NET_TARGET_HEAD的字节数)
                int uintSize = (payloadSizeToDecimal - 64) / targetNum;
@@ -213,7 +232,7 @@
                    // log.info("报警类型:" + alarmType);

                    byte[] szName = Arrays.copyOfRange(data, index + 64, index + 96);
                    int position = findIndexOfDoubleZero(szName);
                    int position = ByteUtils.findIndexOfDoubleZero(szName);
                    String alarmPointName;
                    if (position != -1) {
                        byte[] result = new byte[position];
@@ -242,10 +261,11 @@
                    ardAlarmRadar.setAlarmType(alarmType);
                    radarAlarmInfos.add(ardAlarmRadar);
                }

                //endregion
            }
            if (Arrays.equals(cmdId, new byte[]{0x04})) {
                //String hexString = DatatypeConverter.printHexBinary(data);
                //region抽油机AI状态反馈
                String hexString = DatatypeConverter.printHexBinary(data);
                //log.info(hexString);

                byte[] dwTim = Arrays.copyOfRange(data, 4, 8);
@@ -257,9 +277,11 @@

                byte[] wTargetNum = Arrays.copyOfRange(data, 8, 10);
                wTargetNum = ByteUtils.toLittleEndian(wTargetNum);
                int targetNum = ByteUtils.bytesToDecimal(wTargetNum);
                //log.info("目标总点数(转整型):" + targetNum);

                targetNum = ByteUtils.bytesToDecimal(wTargetNum);
                log.info("目标总点数(转整型):" + targetNum);
                if (targetNum == 0) {
                    return;
                }
                //解析NET_TARGET_UNIT(64是NET_TARGET_HEAD的字节数)
                int uintSize = (payloadSizeToDecimal - 64) / targetNum;
                //log.info("单条报警大小:" + uintSize);
@@ -307,9 +329,9 @@
                    //log.info("目标当前状态:" + binaryString);
                    // 提取第0位值
                    // 使用位运算操作判断第0位是否为1
                    boolean isBitSet = (cStat[0] & 0x01) == 0x00;
                    boolean isB0 = (cStat[0] & 0x01) == 0x00;
                    // 判断提取的值
                    if (isBitSet) {
                    if (isB0) {
                        alarmType = "雷达抽油机停机";
                    } else {
                        continue;
@@ -318,7 +340,7 @@

                    byte[] szName = Arrays.copyOfRange(data, index + 32, index + 64);
                    //log.info("所属告警区域名称:" + DatatypeConverter.printHexBinary(szName));
                    int position = findIndexOfDoubleZero(szName);
                    int position = ByteUtils.findIndexOfDoubleZero(szName);
                    String alarmPointName;
                    if (position != -1) {
                        byte[] result = new byte[position];
@@ -335,6 +357,13 @@
                    ardAlarmRadar.setAlarmType(alarmType);
                    radarAlarmInfos.add(ardAlarmRadar);
                }
                //endregion
            }
            if (StringUtils.isEmpty(alarmTime)) {
                return;
            }
            if (targetNum == 0) {
                return;
            }
            RadarAlarmData radarAlarmData = new RadarAlarmData();
            radarAlarmData.setRadarId(id);
@@ -345,18 +374,6 @@
        } catch (Exception ex) {
            log.error("雷达报文解析异常:" + ex.getMessage());
        }
    }

    /**
     * 找到00的索引位置
     */
    private int findIndexOfDoubleZero(byte[] bytes) {
        for (int i = 0; i < bytes.length - 1; i++) {
            if (bytes[i] == 0x00) {
                return i;
            }
        }
        return -1;
    }

}

 src/main/java/com/ard/utils/udp/NettyUdpServer.java

@@ -18,9 +18,8 @@
    /**
     * 启动服务
     */
    @Async
    @Async("alarm")
    public void init(int port) {

        //表示服务器连接监听线程组，专门接受 accept 新的客户端client 连接
        EventLoopGroup bossLoopGroup = new NioEventLoopGroup();
        try {
@@ -40,7 +39,7 @@
            //7、监听通道关闭事件，应用程序会一直等待，直到channel关闭
            channelFuture.channel().closeFuture().sync();
        } catch (Exception e) {
            log.error("error:"+e.getMessage());
            log.error("error:" + e.getMessage());
        } finally {
            log.info("netty udp close!");
            //8 关闭EventLoopGroup，

 src/main/resources/application.yml

@@ -1,11 +1,34 @@
spring:
  datasource:
      #  使用druid数据库连接池
    druid:
        #开启druid监控web
      stat-view-servlet:
        enabled: true
      driver-class-name: org.postgresql.Driver
          #    url: jdbc:postgresql://111.40.46.199:15432/ry-vue?stringtype=unspecified
          #    username: postgres
          #    password: Yykj.2021
      url: jdbc:postgresql://192.168.1.15:5432/ry-vue?stringtype=unspecified
      username: postgres
      password: postgres
      #最大连接数
      maxActive: 30
      #最小连接数
      minIdle: 5
      #获取连接的最大等待时间
      maxWait: 10000
      #解决mysql8小时的问题
      validation-query: SELECT 'X'
      #空闲连接的检查时间间隔
      timeBetweenEvictionRunsMillis: 60000
      #空闲连接最小空闲时间
      minEvictableIdleTimeMillis: 300000
  netty:
    udp:
      port: 40000
      enabled: true
    tcp:
      host: 112.98.126.2
      port: 1200
      enabled: true
  mqtt:
    host: tcp://192.168.1.15:1883
@@ -18,14 +41,6 @@
    enabled: true
  stealelec:
    url: http://iot.zhdk.net:8090/Warning/GetWarning?userName=cy4oil
  datasource:
    driver-class-name: org.postgresql.Driver
#    url: jdbc:postgresql://111.40.46.199:15432/ry-vue?stringtype=unspecified
#    username: postgres
#    password: Yykj.2021
    url: jdbc:postgresql://192.168.1.15:5432/ry-vue?stringtype=unspecified
    username: postgres
    password: postgres
mybatis:
  typeAliasesPackage: com.ard.alarm.**.domain
  mapperLocations: classpath:/mapper/*.xml

 src/main/resources/logback-spring.xml

@@ -76,13 +76,35 @@
            <onMismatch>DENY</onMismatch>
        </filter>
    </appender>
    <!--雷达报警日志输出-->
    <appender name="radar" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <file>${log.path}/radar.log</file>
    <!--Netty日志输出-->
    <appender name="netty" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <file>${log.path}/netty.log</file>
        <!--循环政策：基于时间创建日志文件-->
        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
            <!--日志文件名格式-->
            <fileNamePattern>${log.path}/radar.%d{yyyy-MM-dd}.log</fileNamePattern>
            <fileNamePattern>${log.path}/netty.%d{yyyy-MM-dd}.log</fileNamePattern>
            <!--日志最大的历史60天-->
            <maxHistory>60</maxHistory>
        </rollingPolicy>
        <encoder>
            <pattern>${log.pattern}</pattern>
        </encoder>
        <filter class="ch.qos.logback.classic.filter.LevelFilter">
            <!--过滤的级别-->
            <level>INFO</level>
            <!--匹配时的操作：接收（记录）-->
            <onMatch>ACCEPT</onMatch>
            <!--不匹配时的操作：拒绝（不记录）-->
            <onMismatch>DENY</onMismatch>
        </filter>
    </appender>
    <!--Mqtt日志输出-->
    <appender name="mqtt" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <file>${log.path}/mqtt.log</file>
        <!--循环政策：基于时间创建日志文件-->
        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
            <!--日志文件名格式-->
            <fileNamePattern>${log.path}/mqtt.%d{yyyy-MM-dd}.log</fileNamePattern>
            <!--日志最大的历史60天-->
            <maxHistory>60</maxHistory>
        </rollingPolicy>
@@ -106,16 +128,20 @@
        <appender-ref ref="sys-info"/>
    </root>
    <!--管线泄露报警操作日志-->
    <root level="INFO">
    <logger name="tube" level="INFO">
        <appender-ref ref="tube"/>
    </root>
    </logger>
    <!--通用光电报警操作日志-->
    <root level="INFO">
    <logger name="camera" level="INFO">
        <appender-ref ref="camera"/>
    </root>
    <!--雷达报警日志-->
    <root level="INFO">
        <appender-ref ref="radar"/>
    </root>
    </logger>
    <!--Netty日志-->
    <logger name="netty"  level="INFO">
        <appender-ref ref="netty"/>
    </logger>
    <!--Mqtt日志-->
    <logger name="mqtt"  level="INFO">
        <appender-ref ref="mqtt"/>
    </logger>
</configuration>