关于Netty使用中黏包拆包带来报错问题及解决

文章目录

    • 问题现象
    • 解决
    • 总结

问题现象

  • 业务场景:雷达作为客户端,平台作为服务端,采用TCP/IP协议的socket连接,数据包采用字节的二进制数据传输
  • 平台与雷达的通信和数据解析,在我接手时,已经开发完成,我接手负责维护和继续开发
  • 由于日常工作忙于大数据分析和指标计算,未对使用正常的通信代码做什么改动,在客户现场使用后发现了一些问题(公司测试没问题)
  • 问题1:雷达实时监测过车车辆,将数据通过socket上报给平台。但平台存入的过车,与现场视频相比,数量上明显偏少,怀疑存在数据丢失现象
  • 问题2:程序刚启动时,产生以下报错,后续也有偶尔类似报错
2022-12-19 10:40:57.215 ERROR 1 --- [ntLoopGroup-3-9] c.n.s.listener.DefaultExceptionListener  : Can't parse 115

java.lang.IllegalArgumentException: Can't parse 115
	at com.newatc.socketio.protocol.PacketType.valueOfInner(PacketType.java:63)
	at com.newatc.socketio.protocol.PacketDecoder.decodePackets(PacketDecoder.java:21)
	at com.newatc.socketio.handler.InPacketHandler.channelRead0(InPacketHandler.java:66)
	at com.newatc.socketio.handler.InPacketHandler.channelRead0(InPacketHandler.java:33)
	at io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:99)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
	at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:302)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
	at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
	at com.newatc.socketio.handler.AuthorizeHandler.channelRead(AuthorizeHandler.java:149)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
	at io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:102)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
	at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:324)
	at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:296)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
	at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
	at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
	at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:719)
	at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:655)
	at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:581)
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493)
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	at java.base/java.lang.Thread.run(Unknown Source)
	

解决

  • 针对问题2,追踪业务代码,发现是解析到的数据包类型出错,不在规定范围内。由于我们的平台和雷达设备都在客户内网内(不通互联网),基本不会有第三方数据侵入,数据上报类型不应该超出规定范围,肯定是平台数据接收解析出了问题
  • 针对问题1,用了另一个工具(雷达上位机软件)接收雷达数据,发现车辆数量正常,初步定位为平台处理socket上报信息时丢失了部分数据
  • 针对问题1,是在客户现场实施时发现的问题,在公司测试环境接入雷达测试时没有问题。对比分析了下两个场景,发现客户的路口数量较多,雷达数量也较多,同时这些路口实时过车也较多。
  • 雷达监测到过车就会触发数据上报,当数据量较大时,单个数据包超过最大长度就会进行拆包,分多个数据包上报
  • 查看了平台已有代码,使用的是Netty,但在数据接收时,没有进行包头校验、长度校验。对于拆包后的数据包(前面几个字节不是包头),跳过包头长度直接进行包类型校验会失败报错(此字节位置可能是负载的实际数据内容,而不是包头包类型),出错就丢掉数据,就会导致上面2个问题
    在这里插入图片描述
  • 先看下优化前,有问题的代码,很粗糙,很明显,但当过车较少,单个数据包包含所有信息时没问题
    // 数据接收经过的一个解码器 InPacketHandler 的接收处理方法
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, PacketsMessage message) throws Exception {
        ByteBuf content = message.getContent();
        ClientHead client = message.getClient();

        if (log.isTraceEnabled()) {
            log.trace("In message: {} sessionId: {}", content.toString(CharsetUtil.UTF_8), client.getSessionId());
        }
        while (content.isReadable()) {
            try {
                Packet packet = decoder.decodePackets(content, client);
                if(packet == null)
                    continue;
                Namespace ns = namespacesHub.get(packet.getNsp());

                NamespaceClient nClient = client.getChildClient(ns);
                if (nClient == null) {
                    log.debug(
                        "Can't find namespace client in namespace: {}, sessionId: {} probably it was disconnected.",
                        ns.getName(),
                        client.getSessionId()
                    );
                    return;
                }
                packetListener.onPacket(packet, nClient);
            } catch (Exception ex) {
                String c = content.toString(CharsetUtil.UTF_8);
                log.error("Error during data processing. Client sessionId: " + client.getSessionId() + ", data: " + c, ex);
                throw ex;
            }
        }
        // 读取完成后,释放ByteBuf,防止内存溢出LEAK
        content.release();
    }
    
    // 上面channelRead0方法里对应的具体解码方法
    public Packet decodePackets(ByteBuf buf, ClientHead client) {
        // 前七个字节为 : 标志(4)-负载长度(2)-协议版本号(1),未校验,直接跳过了
        buf.skipBytes(7);
        Packet packet = null;
        // 第8个字节是包类型
        short pType = buf.readUnsignedByte();
        PacketType packetSubType = PacketType.valueOfInner(pType);
        // 第 9、10字节是校验位和预留位,也跳过了
        buf.skipBytes(2);
        // 这里也没进行负载长度校验,只校验了是否还有可读字节
        if (buf.readableBytes() < 1){
            // 直接返回了,已读的字节也没有回滚回去,这样后面数据包上报依然不完整,解析依然会有问题
            return packet;
        }
        short oID = buf.readUnsignedByte();
        ObjectId objectId = ObjectId.valueOf(oID);

        switch (packetSubType) {
            case QUERY_RESULT:
                break;
            case REPLY:
                break;
            case REPORT:
                packet = new Packet(PacketType.MESSAGE);
                packet.setSubType(packetSubType);
                packet.setObjectId(objectId);

                switch (objectId){
                    case REALTIME_DATA:
                        packet.setName(EventName.REALTIME_DATA);
                        break;
                    case PASSING_VEHICLE:
                        packet.setName(EventName.PASSING_VEHICLE);
                        break;
                    case TRAFFIC_STATUS:
                        packet.setName(EventName.TRAFFIC_STATUS);
                        break;
                    case TRAFFIC_STATS:
                        packet.setName(EventName.FLOW_STATS);
                        break;
                    case PERFORMANCE:
                        packet.setName(EventName.PERFORMANCE);
                        break;
                    case TRAFFIC_EVENT:
                        packet.setName(EventName.TRAFFIC_EVENT);
                        break;
                    case RADAR_FAULT:
                        packet.setName(EventName.RADAR_FAULT);
                        break;
                }
                // 解析数据包里负载的数据体
                Object o = readReportObject(objectId, buf);
                List<Object> args = new ArrayList<>();
                args.add(o);
                packet.setData(args);
                break;
            case HEART_BEAT:
                packet = new Packet(PacketType.PING);
                packet.setSubType(packetSubType);
                packet.setObjectId(ObjectId.HEART_BEAT_FROM_RADAR);

                break;
            default:
                break;
        }

        return packet;
    }

  • 优化后的代码,主要做了包头校验和负载长度校验
  • 如果不是包头则舍弃,直到读到包头开始解析,这是为了防止程序启动时接收到的就是一个不完整包
  • 读取完包头后,如果剩余字节长度小于负载长度,回滚游标到包头位置,返回,继续等待下一个数据包上报再一起解析
    public Packet decodePackets(ByteBuf buf, ClientHead client) {
        // 将起始位置记录下,后面如果不完整,读取失败,回复到此位置
        int savedReaderIndex = buf.readerIndex();

        // 前七个字节为 : 标志CYRC(4)-负载长度(2)-协议版本号(1),除了负载长度,其他没有用,直接跳过
        // 读取前四个字节,如果不是包头,则继续往下读
        byte head1 = buf.readByte();
        if (head1 != 0x43) {
            logger.info("第一个字符不是 C ,舍弃");
            return null;
        }
        byte head2 = buf.readByte();
        if (head2 != 0x59) {
            logger.info("第二个字符不是 Y ,舍弃");
            return null;
        }
        byte head3 = buf.readByte();
        if (head3 != 0x52) {
            logger.info("第三个字符不是 R ,舍弃");
            return null;
        }
        byte head4 = buf.readByte();
        if (head4 != 0x43) {
            logger.info("第四个字符不是 C ,舍弃");
            return null;
        }
        // 读取负载长度
        int dataLength = buf.readUnsignedShort();
        buf.skipBytes(1);
        // 第8个字节为包类型
        short pType = buf.readUnsignedByte();
        PacketType packetSubType = PacketType.valueOfInner(pType);
        // 第9/10字节,为校验位(1)-Reserve(1),暂未实际使用,直接跳过了
        buf.skipBytes(2);
        // 后续数据长度小于负载长度,则等待后面的数据上报再一起解析
        if (buf.readableBytes() < dataLength) {
            buf.readerIndex(savedReaderIndex);
            logger.info("剩余消息长度,小于负载(消息内容)长度 ,重置游标返回,等待后面数据上报一起解析");
            return null;
        }
        // 包头十个字节,后面为负载(消息内容),负载的第一个字节为对象标识,剩余为对象数据内容
        Packet packet = null;
        short oID = buf.readUnsignedByte();
        ObjectId objectId = ObjectId.valueOf(oID);

        switch (packetSubType) {
            case QUERY_RESULT:
                break;
            case REPLY:
                break;
            case REPORT:
                packet = new Packet(PacketType.MESSAGE);
                packet.setSubType(packetSubType);
                packet.setObjectId(objectId);

                switch (objectId) {
                    case REALTIME_DATA:
                        packet.setName(EventName.REALTIME_DATA);
                        break;
                    case PASSING_VEHICLE:
                        packet.setName(EventName.PASSING_VEHICLE);
                        break;
                    case TRAFFIC_STATUS:
                        packet.setName(EventName.TRAFFIC_STATUS);
                        break;
                    case TRAFFIC_STATS:
                        packet.setName(EventName.FLOW_STATS);
                        break;
                    case PERFORMANCE:
                        packet.setName(EventName.PERFORMANCE);
                        break;
                    case TRAFFIC_EVENT:
                        packet.setName(EventName.TRAFFIC_EVENT);
                        break;
                    case RADAR_FAULT:
                        packet.setName(EventName.RADAR_FAULT);
                        break;
                }

                Object o = readReportObject(objectId, buf);
                List<Object> args = new ArrayList<>();
                args.add(o);
                packet.setData(args);
                break;
            case HEART_BEAT:
                packet = new Packet(PacketType.PING);
                packet.setSubType(packetSubType);
                packet.setObjectId(ObjectId.HEART_BEAT_FROM_RADAR);

                break;
            default:
                break;
        }

        return packet;
    }

总结

在使用Netty进行TCP数据传输时,由于TCP是一个面向流的协议,消息会被拆分成多个字节流进行发送,因此接收方收到消息时,可能会出现黏包和拆包现象。

黏包指的是接收方一次性收到了多个完整的消息,而拆包则是接收方收到了不完整的消息。这种现象的出现是由于TCP是面向流的、无边界的协议,不保留数据报的边界。

为了解决黏包和拆包问题,Netty提供了多种解决方法:

  1. 消息定长:即发送方发送的每个消息长度固定,接收方接收到固定长度的字节流后进行消息的解析,这个是一种简单有效的实现方法。但是存在一个问题,不同消息长度不同,如心跳消息和数据上报消息,消息定长则部分消息必须填充补偿,显得浪费带宽。

  2. 消息分隔符:发送方在每个消息后添加特殊的分隔符,接收方根据分隔符对消息进行解码和拆分。使用这种实现方式时,要确保分隔符是唯一的,不会和消息内容重复,造成错误分隔。

  3. 消息头部标识:发送方在消息头部加入固定长度的表示消息长度的字段,接收方根据消息长度字段对消息进行解码和拆分。这是一种比较通用的实现,消息头部可以加一些标识、负载长度、校验位等,就可以正确识别到包头位置和对数据完整性进行校验。

  4. 基于Netty编解码器:Netty提供了一系列编解码器,可以自定义编解码器来控制消息的长度和格式。我们按照自己的诗句情况进行参数配置后,就可以实现自动的黏包和拆包处理。

总的来说,Netty提供了多种方式来解决黏包和拆包问题,我们可以根据业务需求选择合适的方式进行实现。只要双方约定了通信协议且严格按照协议发送数据,并且代码已经处理了黏包拆包,数据解析应该就没有问题。

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:/a/13540.html

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

如何开发一款用户体验优秀的语音交友app?

在数字时代&#xff0c;人们越来越依赖智能手机上的应用程序来与他人进行交流。其中&#xff0c;语音交友app成为了最受欢迎的应用之一。然而&#xff0c;开发一款成功的语音交友app需要深入了解用户需求与体验。本文将探讨如何开发一款用户体验优秀的语音交友app。 着眼于用户…

文心一格,百度AI作画产品

文章目录 AIGC什么是AI作画&#xff1f;Prompt文心一格使用方法注册账号使用AI绘图 AIGC的未来发展结语 AIGC AIGC&#xff08;AI Generated Content&#xff09;是指利用人工智能生成内容。是利用人工智能来生成你所需要的内容&#xff0c;GC的意思是创作内容。与之相对应的概…

Linux-驱动开发-基础温习

一、裸机开发和驱动开发的区别&#xff1a; 裸机开发&#xff1a;底层&#xff08;相对于linux来说&#xff09;&#xff0c;库 二、linux驱动开发-根据各种框架进行开发 1、 外设比较多&#xff0c;资源多&#xff0c;资料非常少&#xff0c;官方的SDK;直接操作寄存器不显示…

JUC高级十二-ReentrantLock、ReentrantReadWriteLock、StampedLock

无锁→独占锁→读写锁→邮戳锁 1. 关于锁的大厂面试题 你知道Java里面有哪些锁?你说你用过读写锁&#xff0c;锁饥饿问题是什么&#xff1f;有没有比读写锁更快的锁&#xff1f;StampedLock知道吗?(邮戳锁/票据锁)ReentrantReadWriteLock有锁降级机制策略你知道吗&#xff1…

ChatGPT | 申请与使用new bing的实用教程

1. 教程参考&#xff1a; https://juejin.cn/post/7199557716998078522 2.在参考上述教程遇到的问题与解决 2.1 下载dev浏览器的网址打不开 egde dev下载地址&#xff08;上面网站上的&#xff09;我电脑打不开 换用下面的网址即可 https://www.microsoftedgeinsider.com/z…

九种 OOM 常见原因及解决方案(IT枫斗者)

九种 OOM 常见原因及解决方案(IT枫斗者) 什么是OOM&#xff1f; OOM&#xff0c;全称“Out Of Memory”&#xff0c;翻译成中文就是“内存用完了”&#xff0c;来源于java.lang.OutOfMemoryError。看下关于的官方说明&#xff1a;Thrown when the Java Virtual Machine canno…

代码随想录算法训练营第三十六天|435. 无重叠区间、763.划分字母区间 、56. 合并区间

文章目录 重叠问题435. 无重叠区间763.划分字母区间:star:56. 合并区间 重叠问题 这几道题都是判断区间重叠&#xff0c;区别就是判断区间重叠后的逻辑。 435. 无重叠区间 链接:代码随想录 解题思路&#xff1a; 这道题和射气球的题几乎思路一样 不断求出重叠的最小右区间&a…

通俗讲解什么是Socket通讯

Socket通讯原理 1、什么是Socket&#xff1f; Socket&#xff0c;即套接字。就是两台主机之间逻辑连接的端点。&#xff08;通俗来说&#xff1a;网络上的两个程序通过一个双向的通信连接实现数据的交换&#xff0c;这个连接的一端称为一个socket&#xff09;。 Socket是一套…

CSS基础——盒子模型

目录 简介 盒子模型组成 内容区 内边距 边框 border-width border-color border-style border 外边距 负值 auto 简写属性 垂直外边距的重叠 浏览器默认设置 内联元素的盒子 简介 在网页中&#xff0c;一切都是可以看作为“盒子”。 在css处理网页的时候&…

常见的四种排名函数的用法(sql)

四个排名函数&#xff1a; 1.row_number 2.rank 3.dense_rank 4.ntile 1. ROW_NUMBER&#xff08;排名场景推荐&#xff09; 1.1 介绍 在 SQL 中&#xff0c;ROW_NUMBER() 是一个窗口函数&#xff0c;它为结果集中的每一行分配一个唯一的序号。该函数的语法如下&#xff1a; …

内网穿透实现在外远程连接RabbitMQ服务

文章目录 前言1.安装erlang 语言2.安装rabbitMQ3. 内网穿透3.1 安装cpolar内网穿透(支持一键自动安装脚本)3.2 创建HTTP隧道 4. 公网远程连接5.固定公网TCP地址5.1 保留一个固定的公网TCP端口地址5.2 配置固定公网TCP端口地址 转载自远控源码文章&#xff1a;无公网IP&#xff…

srm采购管理系统有那些功能

srm采购管理系统&#xff0c;是通过系统的手段对采购过程进行管理和控制&#xff0c;实现降低成本、提高效益、提高企业核心竞争力的目的。那么 srm采购管理系统有哪些功能呢&#xff1f; 计划管理 srm采购管理系统提供了各种物料需求计划的功能&#xff0c;以帮助企业制定并控…

设置环境变量

文章目录 window设置linux设置python设置 window设置 命令行设置 set 临时设置setx 永久设置 # 打开一个cmd命令行 set # 查看所有环境变量 set FLASK_APPsuperset # 临时设置&#xff0c;当前窗口有效 set FLASK_APP%FLASK_APP%;777 # # 查看 echo %FLASK_APP%# 永久设置…

k8s安装部署apollo配置中心

一、文章大纲 二、安装MySQL5.7 三、创建apollo-config 四、创建apollo-admin 五、创建apollo-portal 六、查看apollo各个组件服务状态 七、访问apollo 八、nginx代理配置转发#注意 一定要先启动apollo-config&#xff0c;再启动apollo-admin&#xff0c;最后启动apollo-porta…

matrix部署

一、环境描述 首先matrix是一个去中心化的聊天服务&#xff0c;matrix实现了端对端的加密&#xff0c;这意味着不仅其他人无法查看你的聊天内容&#xff0c;哪怕你更换了一个终端&#xff0c;你也需要私钥才能够查看你的聊天记录。 这是终极的隐私保护方案&#xff0c;因为一旦…

深入理解机器学习——过拟合(Overfitting)与欠拟合(Underfitting)

分类目录&#xff1a;《深入理解深度学习》总目录 机器学习的主要挑战是我们的算法必须能够在先前未观测的新输入上表现良好&#xff0c;而不只是在训练集上表现良好。在先前未观测到的输入上表现良好的能力被称为泛化&#xff08;Generalization&#xff09;。通常情况下&…

20、单元测试

文章目录 1、JUnit5 的变化2、JUnit5常用注解3、断言&#xff08;assertions&#xff09;1、简单断言2、数组断言3、组合断言4、异常断言5、超时断言6、快速失败 4、前置条件&#xff08;assumptions&#xff09;5、嵌套测试6、参数化测试7、迁移指南 【尚硅谷】SpringBoot2零基…

医院体检PEIS系统源码

一、医院体检系统概述 1. 医院体检系统概述 目前&#xff0c;大多数的体检还停留在手工操作上&#xff0c;如单位体检时手工书写体检人员信息、医生手工书写体检结果、检验报告打印后进行手工粘贴等&#xff0c;这样造成极大的工作量&#xff0c;效率低下&#xff0c;而且极易…

OpenCV 安卓编程示例:1~6 全

原文&#xff1a;OpenCV Android Programming By Example 协议&#xff1a;CC BY-NC-SA 4.0 译者&#xff1a;飞龙 本文来自【ApacheCN 计算机视觉 译文集】&#xff0c;采用译后编辑&#xff08;MTPE&#xff09;流程来尽可能提升效率。 当别人说你没有底线的时候&#xff0c;…

npm和yarn的相同点和不同点

官网 npmhttps://www.npmjs.com Home | Yarn - Package ManagerFast, reliable, and secure dependency management.https://yarnpkg.com Fast, disk space efficient package manager | pnpmFast, disk space efficient package managerhttps://pnpm.io 使用场景 npm&#x…