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; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; import org.apache.mina.core.session.IoSession; public class MyThreadJob implements ThreadPool.Job { private static final Logger log = LogManager.getLogger(MyThreadJob.class) ; public String rtuAddr; public String serverIp; public Integer serverPort; public IoSession session ; public boolean exceptionOnConnect = false ; public static final int connectTimeout = 3000 ; public int sendTimes = 1 ;//发送数据次数 public int heartbeatTimes = 1 ;//上报心跳次数 public long overStart = 0L; public boolean isOver = false ; public MyThreadJob(){ } public MyThreadJob(String rtuAddr, String serverIp, Integer serverPort){ this.rtuAddr = rtuAddr ; this.serverIp = serverIp ; this.serverPort = serverPort ; } @Override public void execute() throws Exception { if(!this.isOver){ if(session != null){ if(sendTimes <= ServerProperties.sendTimes){ log.info("RTU" + rtuAddr + "开始任务"); log.info("RTU" + rtuAddr + "将要执行" + ServerProperties.sendTimes + "轮次任务，当前轮次是" + sendTimes); sendDataOfP206V1_0_0() ; }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(){ try{ if(heartbeatTimes > ServerProperties.heartbeatTimes){ heartbeatTimes = 1 ; this.sendReportData() ; TcpClUnit.clientSendData(); sendTimes++ ; }else{ this.sendHeartbeat() ; TcpClUnit.clientSendData(); heartbeatTimes++ ; } }catch(Exception e){ e.printStackTrace(); } } private void sendHeartbeat(){ try{ byte[] bs = this.createHeartbeat() ; this.session.write(bs) ; log.info("RTU" + rtuAddr + "发送了心跳数据：" + ByteUtil.bytes2Hex(bs, false)); }catch (Exception e){ e.printStackTrace(); } } 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(){ 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 ; } @Override public void destroy() { } @Override public boolean isDestroy() { return false; } }