增加水肥机、气象站、墒情站实体及相关代码。

liurunyu

2025-06-18 3ca5d816ee19dd18213b9be99d6668ed6a83c627

 pipIrr-platform/pipIrr-mw/pipIrr-mwTest-client/src/main/java/com/dy/testClient/tcpClient/MyThreadJob.java

@@ -1,8 +1,5 @@
package com.dy.testClient.tcpClient;

import com.dy.common.mw.protocol.p206V1_0_0.CommonV1_0_1;
import com.dy.common.mw.protocol.p206V1_0_0.ProtocolConstantV206V1_0_0;
import com.dy.common.mw.protocol.p206V1_0_0.parse.global.GlCreate;
import com.dy.common.threadPool.ThreadPool;
import com.dy.common.util.ByteUtil;
import com.dy.testClient.ServerProperties;
@@ -14,6 +11,16 @@

    private static final Logger log = LogManager.getLogger(MyThreadJob.class) ;

    public static String icCardAddr = "3B7D1E1A" ;
    public static String icCardNo = "61181622830147821" ;
    public static String openValveDt = "240711163700" ;
    public static String cdC0 = "C0" ;
    public static String cd83Open = "83Open" ;
    public static String cd83Close = "83Close" ;


    public static String curCd = cdC0 ;//通过修改这个值，来改变上报数据类型

    public String rtuAddr;
    public String serverIp;
    public Integer serverPort;
@@ -23,9 +30,10 @@

    public static final int connectTimeout = 3000 ;

    public int sendTimes = 0 ;//发送数据次数
    public int heartbeatTimes = 0 ;//上报心跳次数
    public int sendTimes = 1 ;//发送数据次数
    public int heartbeatTimes = 1 ;//上报心跳次数

    public long overStart = 0L;
    public boolean isOver = false ;

    public MyThreadJob(){
@@ -38,22 +46,45 @@

    @Override
    public void execute() throws Exception {
        log.info("RTU" + rtuAddr + "开始任务");
        if(session != null){
            log.info("RTU" + rtuAddr + "将要执行" + ServerProperties.sendTimes + "轮次任务，当前轮次是" + sendTimes);
            if(sendTimes <= ServerProperties.sendTimes){
                sendDataOfP206V1_0_0() ;
            }else{
                this.jobOver() ;
        if(!this.isOver){
            if(session != null){
                if(sendTimes <= ServerProperties.sendTimes){
                    log.info("RTU" + rtuAddr + "开始任务");
                    log.info("RTU" + rtuAddr + "将要执行" + ServerProperties.sendTimes + "轮次任务，当前轮次是" + sendTimes);
                    sendDataOfP206V1(curCd) ;
                }else{
                    if(ServerProperties.sendOverThenCloseConnect == 1){
                        log.info("RTU" + rtuAddr + "等待一会，以接收通信中间件下行数据");
                        if(overStart == 0){
                            overStart = System.currentTimeMillis() ;
                        }else{
                            long now = System.currentTimeMillis() ;
                            if(now - overStart >= 30 * 1000){
                                this.jobOver() ;
                            }
                        }
                    }else{
                        this.jobOver() ;
                    }
                }
            }
        }
    }

    private void sendDataOfP206V1_0_0(){
    private void sendDataOfP206V1(String cd){
        try{
            if(heartbeatTimes >= ServerProperties.heartbeatTimes){
                heartbeatTimes = 0 ;
                this.sendReportData() ;
            if(heartbeatTimes > ServerProperties.heartbeatTimes){
                heartbeatTimes = 1 ;
                byte[] bs = null ;
                if(cd.equals(cdC0)){
                    bs = RtuUpData.createCdC0(this.rtuAddr) ;
                }else if(cd.equals(cd83Open)){
                    bs = RtuUpData.createCd83Open(this.rtuAddr, icCardAddr, icCardNo, openValveDt) ;
                }else if(cd.equals(cd83Close)){
                    bs = RtuUpData.createCd83Close(this.rtuAddr, icCardAddr, icCardNo, openValveDt) ;
                }
                this.session.write(bs) ;
                log.info("RTU" + rtuAddr + "发送了上行数据：" + ByteUtil.bytes2Hex(bs, false));
                TcpClUnit.clientSendData();
                sendTimes++ ;
            }else{
@@ -67,7 +98,7 @@
    }
    private void sendHeartbeat(){
        try{
            byte[] bs = this.createHeartbeat() ;
            byte[] bs = RtuUpData.createHeartbeat(this.rtuAddr) ;
            this.session.write(bs) ;
            log.info("RTU" + rtuAddr + "发送了心跳数据：" + ByteUtil.bytes2Hex(bs, false));
        }catch (Exception e){
@@ -75,144 +106,15 @@
        }
    }

    private void sendReportData(){
        try{
            byte[] bs = this.createReport() ;
            this.session.write(bs) ;
            log.info("RTU" + rtuAddr + "发送了上报数据：" + ByteUtil.bytes2Hex(bs, false));
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    private void jobOver(){
        session.closeOnFlush() ;
        if(ServerProperties.sendOverThenCloseConnect == 1){
            session.closeOnFlush() ;
        }
        this.isOver = true ;
        TcpClUnit.clientOver() ;
    }

    /**
     * 构造心跳数据
     * @return 字节数组
     * @throws Exception 异常
     */
    public byte[] createHeartbeat( ) throws Exception {
        CommonV1_0_1 commonV1_0_1 = new CommonV1_0_1() ;
        byte[] bytes ;
        byte[] bsHead = new byte[ProtocolConstantV206V1_0_0.lenHead2Code] ;
        byte index = 0 ;
        bsHead[index] = ProtocolConstantV206V1_0_0.P_Head_Byte ;

        index++ ;
        bsHead[index] = 0 ;//帧长度

        index++ ;
        bsHead[index] = ProtocolConstantV206V1_0_0.P_Head_Byte ;

        index++ ;
        bsHead[index] = commonV1_0_1.createCtrl((byte)0x80, (byte)0) ;//控制域功能码：确认 认可

        index++ ;
        GlCreate.createRtuAddr(this.rtuAddr, bsHead, index);
        index += 5 ;

        ByteUtil.hex2Bytes("02", bsHead, index) ;

        byte[] bs = new byte[1] ;
        bs[0] = (byte)0xF2 ;//数据域： 1 个字节，F0 登录， F1 退出登录，F2 在线保持。

        bytes = ByteUtil.bytesMerge(bsHead, bs) ;

        GlCreate.createLen(bytes);//长度放字节数组中

        byte[] bsTail = GlCreate.createCrcTail(bytes) ;//CRC和尾叠加字节数组中

        bytes = ByteUtil.bytesMerge(bytes, bsTail) ;

        return bytes ;
    }

    /**
     * 构造上报数据
     * @return 字节数组
     * @throws Exception 异常
     */
    public byte[] createReport( ) throws Exception {
        CommonV1_0_1 commonV1_0_1 = new CommonV1_0_1() ;
        byte[] bytes ;
        byte[] bsHead = new byte[ProtocolConstantV206V1_0_0.lenHead2Code] ;
        byte index = 0 ;
        bsHead[index] = ProtocolConstantV206V1_0_0.P_Head_Byte ;

        index++ ;
        bsHead[index] = 0 ;//帧长度

        index++ ;
        bsHead[index] = ProtocolConstantV206V1_0_0.P_Head_Byte ;

        index++ ;
        bsHead[index] = commonV1_0_1.createCtrl((byte)0x80, (byte)3) ;//控制域功能码3：自报帧，流量（水量）参数

        index++ ;
        GlCreate.createRtuAddr(this.rtuAddr, bsHead, index);
        index += 5 ;

        ByteUtil.hex2Bytes("C0", bsHead, index) ;

        byte[] bs = new byte[2] ;
        bs[0] = (byte)0x04 ;//有流量数据
        bs[1] = (byte)0xE0 ;//有水压、电池电压、信号强度

        bytes = ByteUtil.bytesMerge(bsHead, bs) ;

        bs = new byte[5] ;//瞬时流量
        bs[0] = ByteUtil.int2BCD_LE (21)[0] ;
        bs[1] = ByteUtil.int2BCD_LE (43)[0] ;
        bs[2] = ByteUtil.int2BCD_LE (65)[0] ;
        bytes = ByteUtil.bytesMerge(bytes, bs) ;

        bs = new byte[5] ;//累计流量
        bs[0] = ByteUtil.int2BCD_LE (21)[0] ;
        bs[1] = ByteUtil.int2BCD_LE (43)[0] ;
        bs[2] = ByteUtil.int2BCD_LE (65)[0] ;
        bs[3] = ByteUtil.int2BCD_LE (87)[0] ;
        bytes = ByteUtil.bytesMerge(bytes, bs) ;

        bs = new byte[5] ;//损失流量
        bs[0] = ByteUtil.int2BCD_LE (10)[0] ;
        bs[1] = ByteUtil.int2BCD_LE (32)[0] ;
        bs[2] = ByteUtil.int2BCD_LE (54)[0] ;
        bytes = ByteUtil.bytesMerge(bytes, bs) ;

        bs = new byte[4] ;//水压
        bs[0] = ByteUtil.int2BCD_LE (10)[0] ;
        bs[1] = ByteUtil.int2BCD_LE (32)[0] ;
        bs[2] = ByteUtil.int2BCD_LE (4)[0] ;
        bytes = ByteUtil.bytesMerge(bytes, bs) ;

        bs = new byte[4] ;//电池电压
        bs[0] = ByteUtil.int2BCD_LE (10)[0] ;
        bs[1] = ByteUtil.int2BCD_LE (32)[0] ;
        bs[2] = ByteUtil.int2BCD_LE (10)[0] ;
        bs[3] = ByteUtil.int2BCD_LE (32)[0] ;
        bytes = ByteUtil.bytesMerge(bytes, bs) ;

        bs = new byte[1] ;//信号强度
        bs[0] = ByteUtil.int2BCD_LE (21)[0] ;
        bytes = ByteUtil.bytesMerge(bytes, bs) ;

        bs = new byte[6] ;//时钟
        GlCreate.createTp(bs, 0);
        bytes = ByteUtil.bytesMerge(bytes, bs) ;

        GlCreate.createLen(bytes);//长度放字节数组中

        byte[] bsTail = GlCreate.createCrcTail(bytes) ;//CRC和尾叠加字节数组中

        bytes = ByteUtil.bytesMerge(bytes, bsTail) ;

        return bytes ;
    }