更改雷达tcp客户端

‘liusuyi’

2023-12-28 eae9c75f70004dfe128718c63fe04c1a5cc35b01

更改雷达tcp客户端

 src/main/java/com/ard/utils/netty/tcp/BootNettyChannelInboundHandlerAdapter.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,447 @@
package com.ard.utils.netty.tcp;

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.mqtt.MqttProducer;
import com.ard.utils.util.ByteUtils;
import com.ard.utils.util.GisUtils;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelId;
import io.netty.channel.ChannelInboundHandlerAdapter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import javax.xml.bind.DatatypeConverter;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

import static com.ard.utils.util.ByteUtils.byteToBitString;
import static com.ard.utils.util.ByteUtils.toLittleEndian;

@Slf4j(topic = "netty")
@ChannelHandler.Sharable
public class BootNettyChannelInboundHandlerAdapter extends ChannelInboundHandlerAdapter {
    
    /**
     * ä»ŽæœåŠ¡ç«¯æ”¶åˆ°æ–°çš„数据时，这个方法会在收到消息时被调用
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg1) throws Exception, IOException {
        if(msg1 == null){
            return;
        }

        //System.out.println("channelRead:read msg:"+msg1.toString());
        //BootNettyClientChannel bootNettyClientChannel = BootNettyClientChannelCache.get("clientId:"+ctx.channel().id().toString());
        //if(bootNettyClientChannel != null){
        //    System.out.println("to do");
        //    bootNettyClientChannel.setLast_data(msg1.toString());
        //}
        InetSocketAddress inSocket = (InetSocketAddress) ctx.channel().remoteAddress();
        String host = inSocket.getAddress().getHostAddress();
        int port = inSocket.getPort();
        ArdEquipRadar ardEquipRadar = ClientInitialize.tureConnectMap.get(host+":"+port);
        // å¤„理接收到的消息
        ByteBuf msg= (ByteBuf)msg1;
        byte[] byteArray = new byte[msg.readableBytes()];
        msg.getBytes(msg.readerIndex(), byteArray);
        byte[] bytes = MessageParsing.receiveCompletePacket(byteArray);
        if (bytes != null) {
            processData(ardEquipRadar, bytes);
        }
        //回应服务端
        //ctx.write("I got server message thanks server!");
    }
 
    /**
     * ä»ŽæœåŠ¡ç«¯æ”¶åˆ°æ–°çš„数据、读取完成时调用
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws IOException {
        System.out.println("channelReadComplete");
        ctx.flush();
    }
 
    /**
     * å½“出现 Throwable å¯¹è±¡æ‰ä¼šè¢«è°ƒç”¨ï¼Œå³å½“ Netty ç”±äºŽ IO é”™è¯¯æˆ–者处理器在处理事件时抛出的异常时
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws IOException {
//        System.out.println("exceptionCaught");
        cause.printStackTrace();
        ctx.close();//抛出异常，断开与客户端的连接
    }
 
    /**
     * å®¢æˆ·ç«¯ä¸ŽæœåŠ¡ç«¯ç¬¬ä¸€æ¬¡å»ºç«‹è¿žæŽ¥æ—¶ æ‰§è¡Œ
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        // å®¢æˆ·ç«¯ä¸ŽæœåŠ¡ç«¯ å»ºç«‹è¿žæŽ¥
        InetSocketAddress inSocket = (InetSocketAddress) ctx.channel().remoteAddress();
        String clientIp = inSocket.getAddress().getHostAddress();
        int port = inSocket.getPort();
        log.debug("连接成功：【"+clientIp+":"+port+"】");
    }
 
    /**
     * å®¢æˆ·ç«¯ä¸ŽæœåŠ¡ç«¯ æ–­è¿žæ—¶ æ‰§è¡Œ
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception{
        super.channelInactive(ctx);
        InetSocketAddress ipSocket = (InetSocketAddress) ctx.channel().remoteAddress();
        int port = ipSocket.getPort();
        String host = ipSocket.getHostString();
        log.error("与设备" + host + ":" + port + "连接断开!");
        // é‡è¿ž
        BootNettyClientThread thread = new BootNettyClientThread(host,port);
        thread.start();
    }
    /**
     * è§£æžæŠ¥è­¦æ•°æ®
     */
    public void processData(ArdEquipRadar ardEquipRadarbyte, byte[] data) {
        try {
            String radarId = ardEquipRadarbyte.getId();
            String radarName = ardEquipRadarbyte.getName();
            Double radarLongitude = ardEquipRadarbyte.getLongitude();
            Double radarLagitude = ardEquipRadarbyte.getLatitude();
            Double radarAltitude = ardEquipRadarbyte.getAltitude();
            //region crc校验-目前仅用于显示校验结果
            Boolean crc32Check = MessageParsing.CRC32Check(data);
            if (!crc32Check) {
                log.debug("CRC32校验不通过");
            } else {
                //log.debug("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
            String cmdIdStr = DatatypeConverter.printHexBinary(cmdId);
            //log.info("命令ID:" + DatatypeConverter.printHexBinary(cmdId));
            byte[] payloadSize = Arrays.copyOfRange(data, 2, 4);//有效负载大小
            payloadSize = toLittleEndian(payloadSize);
            //log.info("payloadSize:" + DatatypeConverter.printHexBinary(payloadSize));
            int payloadSizeToDecimal = ByteUtils.bytesToDecimal(payloadSize);
            // log.info("有效负载大小(转整型):" + payloadSizeToDecimal);
            //endregion
            List<ArdAlarmRadar> radarAlarmInfos = new ArrayList<>();
            ArdAlarmRadar radarFollowInfo = null;
            //抽油机状态雷达推送集合
            List<ArdAlarmRadar> well = new ArrayList<>();
            String alarmTime = "";
            Integer targetNum = 0;
            log.debug("Processing radar data ã€" + radarName + "】数据-->命令ID:"  + cmdIdStr + "二进制:" + byteToBitString(cmdId[0]));
            //雷达移动防火报警
            if (Arrays.equals(cmdId, new byte[]{0x01})) {
                //region å‘Šè­¦ä¿¡æ¯åé¦ˆ
                byte[] dwTim = Arrays.copyOfRange(data, 4, 8);
                dwTim = 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 = toLittleEndian(wTargetNum);
                targetNum = ByteUtils.bytesToDecimal(wTargetNum);
                if (targetNum == 0) {
                    return;
                }
                //log.debug("目标总点数(转整型):" + 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 = toLittleEndian(dwID);
                    int targetId = ByteUtils.bytesToDecimal(dwID);
                    // log.info("目标ID号：" + targetId);

                    byte[] iDistance = Arrays.copyOfRange(data, index + 8, index + 12);
                    iDistance = toLittleEndian(iDistance);
                    double Distance = ByteUtils.bytesToDecimal(iDistance);
                    //log.debug("目标当前直线距离(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);
                    log.info("原始状态:" + byteToBitString(cStat[0]));
                    // cStat = toLittleEndian(cStat);
                    // æå–第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);
                    String alarmPointName = ByteUtils.bytesToStringZh(szName);
                    // log.info("所属告警区域名称:" + alarmPointName);
                    byte[] afTx = Arrays.copyOfRange(data, index + 96, index + 100);
                    afTx = toLittleEndian(afTx);
                    float fTx = ByteUtils.bytesToFloat(afTx);
                    //  log.info("水平角度:" + fTx);
                    byte[] afTy = Arrays.copyOfRange(data, index + 112, index + 116);
                    afTy = toLittleEndian(afTy);
                    float fTy = ByteUtils.bytesToFloat(afTy);
                    //log.debug("垂直角度:" + fTy);
                    // å°†è§’度转换为弧度
                    double thetaRadians = Math.toRadians(fTy + 90);
                    // ä½¿ç”¨æ­£å¼¦å‡½æ•°è®¡ç®—对边长度
                    Distance = Math.sin(thetaRadians) * Distance;
                    //log.debug("目标投影距离(m):" + Distance);

                    Double[] radarXY = {radarLongitude, radarLagitude};
                    Double[] alarmXY = GisUtils.CalculateCoordinates(radarXY, Distance, (double) fTx);
                    log.debug("报警信息：" + "【radarName】" + radarName + "【targetId】" + targetId + "【alarmType】" + alarmType + "【alarmTime】" + alarmTime + "【name】" + alarmPointName);
                    ArdAlarmRadar ardAlarmRadar = new ArdAlarmRadar();
                    ardAlarmRadar.setTargetId(targetId);
                    ardAlarmRadar.setName(alarmPointName);
                    ardAlarmRadar.setLongitude(alarmXY[0]);
                    ardAlarmRadar.setLatitude(alarmXY[1]);
                    ardAlarmRadar.setAlarmType(alarmType);
                    radarAlarmInfos.add(ardAlarmRadar);

                    int bit1 = (cStat[0] >> 1) & 0x1;
                    //目标的B1=1 é”å®š
                    if (bit1 == 1) {
                        radarFollowInfo = ardAlarmRadar;
                        //将追踪锁定的报警对象属性复制给radarFollowInfo对象
                        //BeanUtils.copyProperties(ardAlarmRadar, radarFollowInfo);
                    }
                }
                //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);
                MqttProducer.publish(2, false, "radar", JSON.toJSONString(radarAlarmData));
                if (radarFollowInfo != null) {
                    //当前雷达扫描存在引导跟踪数据，只保留最后一次锁定的数据
                    MqttProducer.publish(2, false, "radarFollowGuide", JSON.toJSONString(radarFollowInfo));
                }
                //抽油机状态MQTT队列
                radarAlarmData.setArdAlarmRadars(well);
                MqttProducer.publish(2, false, "radarWellData", JSON.toJSONString(radarAlarmData));

            }
            //抽油机AI状态反馈
            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 = 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 = toLittleEndian(wTargetNum);
                targetNum = ByteUtils.bytesToDecimal(wTargetNum);
                //log.debug("目标总点数(转整型):" + 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 = toLittleEndian(dwID);
                    int targetId = ByteUtils.bytesToDecimal(dwID);
                    //log.info("目标ID号：" + targetId);
                    //region ä¸éœ€è¦çš„字段
                    byte[] iTw = Arrays.copyOfRange(data, index + 4, index + 8);
                    iTw = toLittleEndian(iTw);
                    int Tw = ByteUtils.bytesToDecimal(iTw);
                    // log.info("目标当前的像素宽度:" + Tw);

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

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

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

                    byte[] cGrp = Arrays.copyOfRange(data, index + 22, index + 23);
                    cGrp = toLittleEndian(cGrp);
                    int Grp = ByteUtils.bytesToDecimal(cGrp);
                    //log.info("所属组:" + Grp);
                    //endregion
                    String alarmType;
                    //抽油机状态变量
                    String wellType;
                    byte[] cStat = Arrays.copyOfRange(data, index + 23, index + 24);
                    cStat = 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 = "雷达抽油机停机";
                        byte[] szName = Arrays.copyOfRange(data, index + 32, index + 64);
                        //log.info("所属告警区域名称:" + DatatypeConverter.printHexBinary(szName));
                        String alarmPointName = ByteUtils.bytesToStringZh(szName);
                        // log.info("所属告警区域名称:" + alarmPointName);
                        //log.debug("报警信息："+ "【radarName】" + radarName + "【targetId】" + targetId + "【name】" + alarmPointName + "【alarmType】" + alarmType + "【alarmTime】" + alarmTime);
                        ArdAlarmRadar ardAlarmRadar = new ArdAlarmRadar();
                        ardAlarmRadar.setTargetId(targetId);
                        ardAlarmRadar.setName(alarmPointName);
                        ardAlarmRadar.setAlarmType(alarmType);
                        radarAlarmInfos.add(ardAlarmRadar);
                        wellType = "停机";
                    } else {
                        wellType = "运行";
                    }
                    //抽油机状态集合中装入数据
                    byte[] szName = Arrays.copyOfRange(data, index + 32, index + 64);
                    String alarmPointName = ByteUtils.bytesToStringZh(szName);
                    log.debug("报警信息：" + "【radarName】" + radarName + "【targetId】" + targetId + "【alarmType】抽油机状态报警" + "【alarmTime】" + alarmTime + "【name】" + alarmPointName + "【alarmState】" + wellType);
                    ArdAlarmRadar wellAlarm = new ArdAlarmRadar();
                    wellAlarm.setTargetId(targetId);
                    wellAlarm.setName(alarmPointName);
                    wellAlarm.setAlarmType(wellType);
                    well.add(wellAlarm);
                }
                //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);
                MqttProducer.publish(2, false, "radar", JSON.toJSONString(radarAlarmData));
                //抽油机状态MQTT队列
                radarAlarmData.setArdAlarmRadars(well);
                MqttProducer.publish(2, false, "radarWellData", JSON.toJSONString(radarAlarmData));
            }
            //强制引导
            if (Arrays.equals(cmdId, new byte[]{0x02})) {
                //region å‘Šè­¦å‰ç«¯å‘送的强制引导信息
                byte[] iDistance = Arrays.copyOfRange(data, 4, 8);
                iDistance = toLittleEndian(iDistance);
                long distance = ByteUtils.bytesToDecimal(iDistance);
                log.info("目标当前距离(m):" + distance);
                byte[] fTx = Arrays.copyOfRange(data, 8, 12);
                fTx = toLittleEndian(fTx);
                float tx = ByteUtils.bytesToFloat(fTx);
                log.debug("方位:" + tx);
                byte[] fTy = Arrays.copyOfRange(data, 12, 16);
                fTy = toLittleEndian(fTy);
                float ty = ByteUtils.bytesToFloat(fTy);
                if (ty < 0) {
                    ty += 360;
                }
                log.debug("俯仰:" + ty);
                Map<String, Object> forceGuideMap = new HashMap<>();
                forceGuideMap.put("distance", distance);
                forceGuideMap.put("p", tx);
                forceGuideMap.put("t", ty);
                forceGuideMap.put("radarId", radarId);
                log.debug("强制引导信息" + forceGuideMap);
                //endregion
                MqttProducer.publish(2, false, "radarForceGuide", JSON.toJSONString(forceGuideMap));
            }
        } catch (Exception ex) {
            log.error("雷达报文解析异常:" + ex.getMessage());
        }
    }
}

 src/main/java/com/ard/utils/netty/tcp/BootNettyChannelInitializer.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,25 @@
package com.ard.utils.netty.tcp;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelInitializer;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.CharsetUtil;

@ChannelHandler.Sharable
public class BootNettyChannelInitializer<SocketChannel> extends ChannelInitializer<Channel> {
 
    @Override
    protected void initChannel(Channel ch) throws Exception {
 
        ch.pipeline().addLast("encoder", new StringEncoder(CharsetUtil.UTF_8));
        ch.pipeline().addLast("decoder", new StringDecoder(CharsetUtil.UTF_8));
        /**
         * è‡ªå®šä¹‰ChannelInboundHandlerAdapter
         */
        ch.pipeline().addLast(new BootNettyChannelInboundHandlerAdapter());
        
    }
 
}

 src/main/java/com/ard/utils/netty/tcp/BootNettyClient.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,104 @@
package com.ard.utils.netty.tcp;

import com.ard.alarm.radar.domain.ArdEquipRadar;
import com.ard.alarm.radar.service.IArdEquipRadarService;
import com.ard.utils.netty.config.NettyTcpConfiguration;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.nio.channels.SocketChannel;
import java.util.List;

@Slf4j(topic = "netty")
@Component
public class BootNettyClient implements ApplicationRunner {
    @Resource
    IArdEquipRadarService ardEquipRadarService;
    @Resource
    NettyTcpConfiguration nettyTcpConfig;
    static Integer waitTimes = 1;
    static EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
    /**
     * åˆå§‹åŒ–Bootstrap
     */
    public static final Bootstrap getBootstrap(EventLoopGroup group) {
        if (null == group) {
            group = eventLoopGroup;
        }
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_KEEPALIVE, true).handler(new BootNettyChannelInitializer<SocketChannel>());
        return bootstrap;
    }
 
    public void connect( String host,int port) throws Exception {
        log.debug("正在进行连接：【" + host+":"+port+"】");
        eventLoopGroup.shutdownGracefully();
        eventLoopGroup = new NioEventLoopGroup();
        Bootstrap bootstrap = getBootstrap(null);
 
        try {
            bootstrap.remoteAddress(host, port);
            // å¼‚步连接tcp服务端
            ChannelFuture future = bootstrap.connect().addListener((ChannelFuture futureListener) -> {
                final EventLoop eventLoop = futureListener.channel().eventLoop();
                if (futureListener.isSuccess()) {
                    BootNettyClientChannel bootNettyClientChannel = new BootNettyClientChannel();
                    Channel channel = futureListener.channel();
                    String id = futureListener.channel().id().toString();
//                    String id = host;
                    bootNettyClientChannel.setChannel(channel);
                    bootNettyClientChannel.setCode("clientId:" + id);
                    BootNettyClientChannelCache.save("clientId:" + id, bootNettyClientChannel);
                    log.debug("netty client start success=" + id);
                } else {
//                    System.err.println("连接失败，" + waitTimes.toString() + "秒后重新连接:" + host);
                    try {
                        Thread.sleep(waitTimes * 1000);
                    } finally {
                        connect(host,port);
                    }
                }
            });
            future.channel().closeFuture().sync();
        } catch (Exception e) {
            System.err.println("连接异常，" + waitTimes.toString() + "秒后重新连接:" + host);
            try {
                Thread.sleep(waitTimes * 1000);
            } finally {
                connect(host,port);
            }
            e.printStackTrace();
        } finally {
            /**
             * é€€å‡ºï¼Œé‡Šæ”¾èµ„源
             */
            eventLoopGroup.shutdownGracefully().sync();
        }
 
    }
    /**
     * åˆå§‹åŒ–方法
     */
    @Override
    public void run(ApplicationArguments args) {
        if (!nettyTcpConfig.getEnabled()) {
            return;
        }
        List<ArdEquipRadar> ardEquipRadars = ardEquipRadarService.selectArdEquipRadarList(new ArdEquipRadar());
        for (ArdEquipRadar ardEquipRadar : ardEquipRadars) {
            String host = ardEquipRadar.getIp();
            Integer port = Integer.valueOf(ardEquipRadar.getPort());
            log.debug("TCP client try to connect radar【：" + host + ":" + port+"】");
            BootNettyClientThread thread = new BootNettyClientThread(host,port);
            thread.start();
        }
    }
}

 src/main/java/com/ard/utils/netty/tcp/BootNettyClientChannel.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,16 @@
package com.ard.utils.netty.tcp;

import io.netty.channel.Channel;
import lombok.Data;

@Data
public class BootNettyClientChannel {

    //    è¿žæŽ¥å®¢æˆ·ç«¯å”¯ä¸€çš„code
    private String code;

    //    å®¢æˆ·ç«¯æœ€æ–°å‘送的消息内容
    private String last_data;

    private transient volatile Channel channel;
}

 src/main/java/com/ard/utils/netty/tcp/BootNettyClientChannelCache.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,29 @@
package com.ard.utils.netty.tcp;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class BootNettyClientChannelCache {
 
    public static volatile Map<String, BootNettyClientChannel> channelMapCache = new ConcurrentHashMap<String, BootNettyClientChannel>();
 
    public static void add(String code, BootNettyClientChannel channel){
        channelMapCache.put(code,channel);
    }
 
    public static BootNettyClientChannel get(String code){
        return channelMapCache.get(code);
    }
 
    public static void remove(String code){
        channelMapCache.remove(code);
    }
 
    public static void save(String code, BootNettyClientChannel channel) {
        if(channelMapCache.get(code) == null) {
            add(code,channel);
        }
    }
 
 
}

 src/main/java/com/ard/utils/netty/tcp/BootNettyClientThread.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,20 @@
package com.ard.utils.netty.tcp;

public class BootNettyClientThread extends Thread {
 
    private final int port;
    private final String host;

    public BootNettyClientThread(String host,int port){
        this.port = port;
        this.host = host;
    }
 
    public void run() {
        try {
            new BootNettyClient().connect(host,port);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}

 src/main/java/com/ard/utils/netty/tcp/ClientHelper.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,41 @@
package com.ard.utils.netty.tcp;

import com.ard.utils.util.ByteUtils;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.xml.bind.DatatypeConverter;
import java.util.Map;

@Component
@EnableScheduling
public class ClientHelper {

    //  ä½¿ç”¨å®šæ—¶å™¨å‘送心跳
    @Scheduled(cron = "0/3 * * * * ?")
    public void heart_timer() {
        //System.err.println("BootNettyClientChannelCache.channelMapCache.size():" + BootNettyClientChannelCache.channelMapCache.size());
        if (BootNettyClientChannelCache.channelMapCache.size() > 0) {
            for (Map.Entry<String, BootNettyClientChannel> entry : BootNettyClientChannelCache.channelMapCache.entrySet()) {
                BootNettyClientChannel bootNettyChannel = entry.getValue();
                //System.out.println(bootNettyChannel.getCode());
                try {
                    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);
                    // log.debug("发送心跳:" + hexString);
                    //message.writeBytes(heart);
                    bootNettyChannel.getChannel().writeAndFlush(Unpooled.buffer().writeBytes(heart));
                } catch (Exception e) {
                    continue;
                }
            }
        }

    }
}

 src/main/java/com/ard/utils/netty/tcp/DynamicClient.java

@@ -26,7 +26,7 @@
 **/
@Slf4j(topic = "netty")
@Component
public class DynamicClient implements ApplicationRunner {
public class DynamicClient {
    @Resource
    IArdEquipRadarService ardEquipRadarService;
    @Resource
@@ -126,25 +126,27 @@
    /**
     * åˆå§‹åŒ–方法
     */
    @Override

    public void run(ApplicationArguments args) {
        if (!nettyTcpConfig.getEnabled()) {
            return;
        }
        EventLoopGroup group = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .handler(new DynamicClientInitializer());
        //EventLoopGroup group = new NioEventLoopGroup();
        //Bootstrap bootstrap = new Bootstrap();
        //bootstrap.group(group)
        //        .channel(NioSocketChannel.class)
        //        .option(ChannelOption.TCP_NODELAY, true)
        //        .option(ChannelOption.SO_KEEPALIVE, true)
        //        .handler(new DynamicClientInitializer());
        List<ArdEquipRadar> ardEquipRadars = ardEquipRadarService.selectArdEquipRadarList(new ArdEquipRadar());
        for (ArdEquipRadar ardEquipRadar : ardEquipRadars) {
            String host = ardEquipRadar.getIp();
            Integer port = Integer.valueOf(ardEquipRadar.getPort());
            log.debug("TCP client try to connect radar【：" + host + ":" + port+"】");
            // connectServer(ardEquipRadar);//连接每一个雷达服务
            connect(bootstrap, ardEquipRadar);
           // connect(bootstrap, ardEquipRadar);
            BootNettyClientThread thread = new BootNettyClientThread(host,port);
            thread.start();
        }
    }
}

 src/main/java/com/ard/utils/netty/tcp/DynamicClientHandler.java

@@ -44,7 +44,6 @@
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) {
        // å¤„理接收到的数据
        // ...
        Channel channel = ctx.channel();
        ArdEquipRadar ardEquipRadar = DynamicClient.ConnectMap.get(channel);
        // å¤„理接收到的消息
@@ -60,7 +59,7 @@
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        // å¤„理异常
        // ...
        log.error("处理异常");
        log.error("处理异常"+cause.getMessage());
    }

    private ScheduledFuture<?> heartbeatTask;