文档章节

基于NIO的消息路由的实现(四) 服务端通讯主线程(2)断包和粘包的处理

皮鞋铮亮
 皮鞋铮亮
发布于 2015/08/18 21:04
字数 1749
阅读 1748
收藏 52

本来我打算单独开一章,专门说明粘包和断包,但是觉得这个事儿我在做的时候挺头疼的,但是对于别人或许不那么重要,于是就在这里写吧。

那么何谓粘包、何谓断包呢?

  • 粘包:我们知道客户端在写入报文给服务端的时候,首先要将需要写入的内容写入Buffer,以ByteBuffer为例,如果你Buffer定义的足够大,并且你发送的报文足够快,此时就会产生粘包现象,举例来说 你发送一个 报文“ M|A”,然后你有发送了一个“M|B”,如果产生粘包,服务端从缓冲区里面读出的就是“M|AM|B”,这样的字符串;也就是说,客户端的第一条报文和第二条报文粘在了一起。

  • 断包:断包往往是在粘包之后产生的,按照刚才的例子,假设你的缓冲区大小设置为4(当然没人会设置这么小的缓冲区,举例子,凑合看吧),如果你发送的报文足够快,就会产生发送给服务器的报文变为这样:第一个包“M|AM”,第二个包“|B”

在大多数的NIO例子中,均不包括此过程的处理,而且很多的例子中也不会浮现这个情况,甚至程序上线,如果系统压力不大,这样的情况都出现的很少(尤其是断包)。值得庆幸的是,这两种情况,我均重现了,我在客户端不做任何停顿的情况下,for循环发送10万条报文给服务端,当我的缓冲区服务端缓冲区设置为4096,客户端缓冲区设置为1024的时候,出现的频率还是蛮高的,可以加大缓冲区来减少断包的情况发生,但是不能避免,粘包则是必然发生的。

好、我回答在第7小点中提到的问题,为什么要在通讯协议的外层在加上四位?这四位就是用来标记我报文指令的长度的,一旦我知道了这个长度,我就可以根据长度对断包和粘包进行相关的处理。具体代码如下:

/**
     * 处理断包和粘包现象
     *
     * @param socketChannel
     * @param byteBuffer
     */
    private void handlePacket(SocketChannel socketChannel, ByteBuffer byteBuffer) {
        //标记读取缓冲区起始位置
        int location = 0;
        //如果缓冲区从0到limit的数量大于包体大小标记数字
        while (byteBuffer.remaining() > PACKET_HEAD_LENGTH) {
            //包体大小标记
            String strBsize;
            //如果endPacket的字节length大于0,则证明:断包的前一截为包含包头和包体的;
            if (endPacketStr.getBytes().length > 0) {

                String strPacket = endPacketStr.substring(PACKET_HEAD_LENGTH) + new String(byteBuffer.array(), 0, remainBodySize);
                byteBuffer.position(remainBodySize);
                location = remainBodySize;
//                                    if(logger.isDebugEnabled()) {
                logger.info("【断包处理】(包含包体)合并后的报文:" + strPacket + ",缓冲区的position:" + location);
//                                    }
                offerPacket(socketChannel, strPacket);
                //处理完毕,清理断包的前一截,以便于下次使用;
                endPacketStr = "";
                //清理后一截报文的字节数标记;
                remainBodySize = 0;
                continue;
                //如果endBufferStr的字节length大于0,则证明:断包的前一截仅包含包头或包头的一部分,不包含包体;
            } else if (endBufferStr.getBytes().length > 0) {

                strBsize = (new StringBuffer(endBufferStr).append(new
                        String(byteBuffer.array(), location, PACKET_HEAD_LENGTH - endBufferStr.getBytes().length))).toString();

                //移动缓冲区position
                byteBuffer.position(PACKET_HEAD_LENGTH - endBufferStr.getBytes().length);
                location = byteBuffer.position();
                //得到包体大小
                int byteBufferSize = Integer.parseInt(strBsize.trim());
                //进行报文合并,把保存的仅包含包头或包头一部分的前一截与后一截合并
                String strPacket = endBufferStr + (new String(byteBuffer.array(), PACKET_HEAD_LENGTH - endBufferStr.getBytes().length, byteBufferSize));
                byteBuffer.position(location + byteBufferSize);//将缓冲区的位置移动到下一个包体大小标记位置
                location = byteBuffer.position();
                logger.info("【断包处理】(不包含包体)合并后的报文:" + strPacket + ",缓冲区的position:" + location);
                offerPacket(socketChannel, strPacket);
                endBufferStr = "";
                continue;
                //进入正常处理(规范的报文处理,不考虑断包)
            } else {
                strBsize = new String(byteBuffer.array(), location, PACKET_HEAD_LENGTH);
                //移动缓冲区position
                byteBuffer.position(location + PACKET_HEAD_LENGTH);
            }
            if (logger.isDebugEnabled()) {
                logger.debug("收到客户端包体大小:" + strBsize + ",查看position变化:" + byteBuffer.position());
            }
            //得到包体大小
            int byteBufferSize = Integer.parseInt(strBsize.trim());
            //如果从缓冲区当前位置到limit大于包体大小,证明粘包了,进行包体处理。等于则为正常包体,不存在粘包现象。
            if (byteBuffer.remaining() >= byteBufferSize) {

                String strPacket = endBufferStr + (new String(byteBuffer.array(), PACKET_HEAD_LENGTH + location, byteBufferSize));
                byteBuffer.position(location + PACKET_HEAD_LENGTH + byteBufferSize);//将缓冲区的位置移动到下一个包体大小标记位置
                if (logger.isDebugEnabled()) {
                    logger.debug("收到客户端包体内容:" + strPacket + ",2查看position变化:" + byteBuffer.position());
                }
                //将字符串报文封装为类
                offerPacket(socketChannel, strPacket);
                location = byteBuffer.position();//设定读取缓冲区起始位置
                //如果缓冲区当前位置到limit小于包体,证明断包了,进行断包处理
            } else {

                endPacketStr = new String(byteBuffer.array(), location, byteBuffer.limit() - location);
                remainBodySize = Integer.parseInt(endPacketStr.substring(0, PACKET_HEAD_LENGTH).trim()) - endPacketStr.getBytes().length + PACKET_HEAD_LENGTH;
                //已经找到断包前半截,所以把整个buffer的position调整至最后,不再处理。等待新的key进入
                byteBuffer.position(byteBuffer.limit());
                logger.info("处理断包仅包含完整包头的尾部报文,缓冲区位置:" + location + ",缓冲区limit:" + byteBuffer.limit() + ",包含完全包头的剩余字符:" + endPacketStr + ",bodySize:" + remainBodySize);

            }
        }
        //处理仅包含包头前一截的报文;
        if (byteBuffer.remaining() > 0) {

            //缓冲区中剩余的仅包含包头前一截的报文
            endBufferStr = new String(byteBuffer.array(), location, byteBuffer.limit() - location);

            logger.info("处理断包仅包含包头前一截的尾部报文,缓冲区位置:" + location + ",缓冲区limit:" + byteBuffer.limit() + ",不包含完全包头的剩余字符:" + endBufferStr);
            //移动缓冲区指针到最后,代表已经保存了前一截报文,无需再进行处理;
            byteBuffer.position(byteBuffer.limit());
        }
        //我也不知道这是否有用,能不能释放内存资源
        byteBuffer.clear();
    }

这块儿很可能有不合理的地方,因为对于一个接近40岁的程序员来说,逻辑在头脑中已经比较混乱了。我知道要对如下几种情况进行处理:

1、粘包,粘包比较好处理,主要是根据包头的前四位,确定包体的大小,然后移动buffer的位置(position),把整个包读出来放入队列就行了;

2、断包:断包分为两种情况,第一种从包头开始就断了,这是你无法获得包体大小,需要把前面的一截保存起来,就必须等下一个报文来了之后,把他们连在一起,然后再做处理;第二种,已经读到完整的包头,仍然需要把前面一截保存起来,确定后面还有多少,然后再处理;我利用了三个类成员:

//断包处理,前一截包含完整包头;
private String endPacketStr = "";
//断包处理,前一截不包含完整包头;
private String endBufferStr = "";
//断包处理,前一截包含完整包头时,包体的大小标记;
private int remainBodySize = 0;

注意这些类的成员需要在使用后,清空,以便于下次使用,否则就乱套了。这块儿代码,我写完就没再看过,挺费神。如果有人能提供更好地办法,不胜感激。


© 著作权归作者所有

共有 人打赏支持
皮鞋铮亮
粉丝 36
博文 12
码字总数 11603
作品 0
沈阳
私信 提问
加载中

评论(1)

宅男小何
宅男小何
这两种情况我觉得可以定义一条消息的结束标记,比如\r\n,收到这个表示这条消息完毕,如果没收到,就等待继续接收数据。
当然你文章里面说的在header里面指定length,也是可以解决的。
http协议我没记错header里面也可以不要content-length这个,也可以的,当没有这个header的时候估计就是当收到结束标记的时候,表示此次通信消息完毕吧,
基于NIO的消息路由的实现(四) 服务端通讯主线程(1)

一、简单介绍: 服务端通讯主线程是消息路由服务的启动类,其主要作用如下: 1、初始化相关配置; 2、根据配置的ip和port创建tcp服务; 3、接收客户端连接,并给客户端分配令牌; 4、接收客户...

皮鞋铮亮
2015/08/18
0
5
基于NIO的消息路由的实现(一) 前言

一、前言: 已经很久没有碰编码了,大概有9年的时间,日新月异的框架和新东西让我眼花缭乱。之前一直在做web相关的应用。由于项目不大,分布式开发在我编码的那个年代里没有做过,后来走上管...

皮鞋铮亮
2015/08/17
0
16
gecko框架概述

1 gecko概述 最近在研究metaq消息队列,它里面用到的NIO通信框架是gecko,文档是这么描述的 Gecko是一个Java NIO的通讯组件,它在一个轻量级的NIO框架的基础上提供了更高层次的封装和功能。 ...

乒乓狂魔
2015/12/19
421
2
java NIO 处理粘包 断包问题

NIO socket是非阻塞的通讯模式,与IO阻塞式的通讯不同点在于NIO的数据要通过channel放到一个缓存池ByteBuffer中,然后再从这个缓存池中读出数据,而IO的模式是直接从inputstream中read。所以...

Flyer_cao
2016/12/19
122
0
如何使用Netty开发实现高性能的RPC服务器

RPC(Remote Procedure Call Protocol)远程过程调用协议,它是一种通过网络,从远程计算机程序上请求服务,而不必了解底层网络技术的协议。说的再直白一点,就是客户端在不必知道调用细节的...

vshcxl
2017/10/20
0
0

没有更多内容

加载失败,请刷新页面

加载更多

day149-2018-11-16-英语流利阅读-待学习

抖音登陆海外市场,歪果仁作何反应? Lala 2018-11-16 1.今日导读 抖音在国内已经火了有一段时间了,但你知道它已经进军海外,而且火了一把吗?中国版抖音的宣传语是“记录美好生活”,但外国...

飞鱼说编程
29分钟前
3
1
Confluence 6 教程:在 Confluence 中导航

当你对 Confluence 有所了解后,你会发现 Confluence 使用起来非常简单。这个教程主要是针对你使用的 Confluence 界面进行一些说明,同时向你展示在那里可以进行一些通用的任务和操作。 空间...

honeymose
今天
2
0
sed, awk 练习

1. sed打印某行到某行之间的内容 2. sed 转换大小写 将单词首字母转化大写 将所有小写转化大写 3. sed 在某一行最后面添加一个数字 4. 删除某行到最后一行 解析: {:a;N;$!ba;d} :a : 是...

Fc丶
今天
2
0
babel6升级到7,jest-babel报错:Requires Babel "^7.0.0-0", but was loaded with "6.26.3".

自从将前端环境更新到babel7,jest-babel之前是基于babel6的,执行时候就会报:Requires Babel "^7.0.0-0", but was loaded with "6.26.3". 很烦,因为连续帮好几台电脑修复这个问题,所以记...

曾建凯
今天
1
0
探索802.11ax

802.11ax承诺在真实条件下改善峰值性能和最差情况。 如何改善今天的Wi-Fi? 在决定如何改进当前版本以外的Wi-Fi时,802.11ac,IEEE和Wi-Fi联盟调查了Wi-Fi部署和行为,以确定更广泛使用的障碍...

linuxprobe16
今天
2
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部