准备工作：搭建netty

1、util文件中准备好数据结构转换，byte[]数组转为int类型

大端写法，从高往低位读

    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;
    }

2、检查报文首尾是否正确

其中 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;
    }

3、处理幂等性问题

处理msgId重复情况，全局存一个变量，与上一次报文进行比较

4、处理报文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;
    }

5、打印接收报文 ByteBuf msg

        byte[] bytes = new byte[Constants.MEG_LENGTH];
        msg.readBytes(bytes);
        log.info("接收到转发服务原始报文 - {}", HexUtil.encodeHexStr(bytes));

后续只需要处理byte数组即可

6、处理报文逻辑netty handler中做处理

/**
 * @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();
    }
}