大端写法,从高往低位读
public static int bytesToInt(byte[] bytes){
int value;
value = ((bytes[0] & 0xff)<<24)|
((bytes[1] & 0xff)<<16)|
((bytes[2] & 0xff)<<8)|
(bytes[3] & 0xff);
return value;
}
其中 public static final byte MEG_SEPARATION_HEAD = (byte) 0x5c;
public static boolean validate(byte[] message) {
if ((message[Constants.HEAD_BYTE_01_POS] == Constants.MEG_SEPARATION_HEAD) && (message[Constants.HEAD_BYTE_02_POS] == Constants.MEG_SEPARATION_HEAD) && (message[Constants.HEAD_BYTE_03_POS] == Constants.MEG_SEPARATION_HEAD) && (message[Constants.HEAD_BYTE_04_POS] == Constants.MEG_SEPARATION_HEAD)
&& (message[Constants.MEG_SEPARATION_TAIL_POS1] == Constants.MEG_SEPARATION_TAIL) && (message[Constants.MEG_SEPARATION_TAIL_POS2] == Constants.MEG_SEPARATION_TAIL) && (message[Constants.MEG_SEPARATION_TAIL_POS3] == Constants.MEG_SEPARATION_TAIL) && (message[Constants.MEG_SEPARATION_TAIL_POS4] == Constants.MEG_SEPARATION_TAIL)) {
return true;
}
log.error("报文头尾出现问题请排查");
return false;
}
处理msgId重复情况,全局存一个变量,与上一次报文进行比较
public static boolean MsgIdCheck(byte[] message) {
byte[] msgIdBytes = generateTypeInt32(message, Constants.MSG_ID_POS);
if (HexUtil.encodeHexStr(msgIdBytes).equalsIgnoreCase("ffffffff")) {
log.error("Modbus总线通讯故障");
return false;
}
return true;
}
byte[] bytes = new byte[Constants.MEG_LENGTH];
msg.readBytes(bytes);
log.info("接收到转发服务原始报文 - {}", HexUtil.encodeHexStr(bytes));
后续只需要处理byte数组即可
/**
* @author lst
* @date 2023/12/20 14:22
* @return null
*/
@Slf4j
public class WebHandler extends SimpleChannelInboundHandler<ByteBuf> {
private Map<Physics, byte[]> previousMessageFields = new LinkedHashMap<>();
private long lastReceivedTimestamp = System.currentTimeMillis();
private WebClient webClient;
public WebHandler(WebClient webClient) {
this.webClient = webClient;
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
if (log.isInfoEnabled()) {
log.info("连接的Web服务器端地址:" + ctx.channel().remoteAddress());
}
super.channelActive(ctx);
}
@Override
protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
byte[] bytes = new byte[Constants.MEG_LENGTH];
msg.readBytes(bytes);
log.info("接收到转发服务原始报文 - {}", HexUtil.encodeHexStr(bytes));
if (!WebUtil.validate(bytes) || !WebUtil.MsgIdCheck(bytes)) {
// Clear the message buffer to handle potential packet fragmentation
msg.clear();
return;
}
//当前接受报文
Map<Physics, byte[]> currentFields = parseFields(bytes);
// 记录接收时间戳
long currentTimestamp = System.currentTimeMillis();
//比较两次报文
if (!previousMessageFields.isEmpty()) {
WebUtil.compareAndPrintChanges(previousMessageFields, currentFields);
}
//更新报文
previousMessageFields = currentFields;
// 更新上一次接收的时间戳
lastReceivedTimestamp = currentTimestamp;
}
private Map<Physics, byte[]> parseFields(byte[] bytes) {
//解析报文放入map中
Map<Physics, byte[]> map = new LinkedHashMap<>();
// Head部分
map.put(new Physics("HeadByte01", "HeadByte01", HEAD_BYTE_01_POS), WebUtil.generateTypeUInt8(bytes, HEAD_BYTE_01_POS));
// Tail部分
map.put(new Physics("TailByte01", "TailByte01", TAIL_BYTE_01_POS), WebUtil.generateTypeUInt8(bytes, TAIL_BYTE_01_POS));
return map;
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
log.error("运行中与Web服务器断开,重连...");
// 重连
webClient.connect();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
log.error(cause.getMessage());
ctx.close();
}
}