由于blog各种垃圾评论太多,而且本人审核评论周期较长,所以懒得管理评论了,就把评论功能关闭,有问题可以直接qq骚扰我

Netty 四个粘包拆包解码器

JAVA 西门飞冰 60℃
[隐藏]

1.为什么会有粘包拆包?

TCP 是个”流”协议,所谓流,就是没有界限的一串数据。TCP底层并不了解上层业务数据的具体含义,它会根据TCP缓冲区的实际情况进行包的划分,所以在业务上认为,一个完整的包可能会被拆分成多个包进行发送,也有可能把多个小的包封装成一个大的数据包发送,这就是所谓的TCP粘包和拆包问题。

关于粘包和拆包可以参考下图的几种情况:

如果一次请求发送的数据量比较小,没达到缓冲区大小,TCP则会将多个请求合并为同一个请求进行发送,这就形成了粘包问题。

如果一次请求发送的数据量比较大,超过了缓冲区大小,TCP就会将其拆分为多次发送,这就是拆包。

image-20221107104113604

UDP: 是没有粘包和拆包的问题,因为UDP有消息保护边界

2.粘包拆包问题案例

案例:写一个Netty程序,客户端连续发送100个fblinux.com字符给服务端,没有添加粘包拆包解码器的情况下,模拟出现粘包和拆包场景。

service 端代码:

public class AppServerEnDecoder {

    private int port;

    public AppServerEnDecoder(int port){
        this.port = port;
    }

    public void run() throws Exception{

        //配置服务端的线程组
        EventLoopGroup bossGroup = new NioEventLoopGroup();

        EventLoopGroup workGroup = new NioEventLoopGroup();

        try{
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup,workGroup)

                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG,128)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new ServerHandler());
                        }
                    });

            System.out.println("Echo 服务器启动");
            //绑定端口,同步等待成功
            ChannelFuture channelFuture =  serverBootstrap.bind(port).sync();
            //等待服务端监听端口关闭
            channelFuture.channel().closeFuture().sync();

        }finally {
            //优雅退出,释放线程池
            workGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }

    }

    public static void main(String[] args) throws Exception {
        int port = 8080;

        if (args.length > 0) {
            port = Integer.parseInt(args[0]);
        }

        new AppServerEnDecoder(port).run();
    }
}

public class ServerHandler extends ChannelInboundHandlerAdapter {
    private int counter;

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

        ByteBuf buf = (ByteBuf)msg;
        byte[] bytes = new byte[buf.readableBytes()];
        buf.readBytes(bytes);

        String body = new String(bytes,"UTF-8").
                substring(0,bytes.length - System.getProperty("line.separator").length());

        System.out.println("服务端收到消息内容为:" + body + ", 收到消息次数:" + ++counter);

    }


    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

客户端代码:

public class AppClientEnDecoder {
    private String host;
    private int port;

    public AppClientEnDecoder(String host, int port) {
        this.host = host;
        this.port = port;
    }

    public void start() throws Exception {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .remoteAddress(new InetSocketAddress(host, port))
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel ch)throws Exception {
                            ch.pipeline().addLast(new ClientHandler());
                        } });
            //连接到服务端,connect是异步链接,再调用同步方法sync,等待连接成功
            ChannelFuture f = bootstrap.connect().sync();
            //阻塞直到客户端通道关闭
            f.channel().closeFuture().sync();
        } finally {
            //优雅退出,释放NIO线程组
            group.shutdownGracefully().sync();
        }
    }


    public static void main(String[] args) throws Exception {
        new AppClientEnDecoder("127.0.0.1", 8080).start();
    }
}

public class ClientHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {

        ByteBuf mes = null;
        byte [] req = ("fblinux.com"+System.getProperty("line.separator")).getBytes();
        //连续发送
        for(int i=0; i< 100; i++){
            mes = Unpooled.buffer(req.length);
            mes.writeBytes(req);
            ctx.writeAndFlush(mes);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {

        cause.printStackTrace();
        ctx.close();
    }
}

服务端输出结果:客户端发送100次请求,服务端仅收到了2次,这种情况就是粘包,出现数据被截断的情况就是拆包。

image-20221105202407835

3.粘包拆包解决方案

发送方:可以关闭Nagle算法(Nagle算法:多个小包集合到一起发送),但是会影响性能

接收方: TCP是无界的数据流,并没有处理粘包现象的机制, 且协议本身无法避免粘包,半包读写的发生需要在应用层进行处理

应用层解决半包读写的办法

1)设置定长消息 (10字符),但是消息单一,不够灵活

2)设置消息的边界 ($$ 切割)

3)使用带消息头的协议,消息头存储消息开始标识及消息的长度信息 Header+Body

Netty提供了四种粘包拆包解决方案:

FixedLengthFrameDecoder:定长解码器,按固定长度进行消息的读取

LineBasedFrameDecoder:行解码器,按行(\r\n)进行消息的读取

DelimiterBasedFrameDecoder:分隔符解码器,按照特殊的分隔符作为消息分隔符进行消息的读

LengthFieldBasedFrameDecoder:自定义长度解码器,通过在消息头中定义消息长度字段来标志消息体的长度,然后根据消息的总长度来读取消息

Netty 无论是服务端还是客户端,当我们读取或者发送消息的时候,都需要考虑TCP底层的粘包/拆包机制。

4.解码器之定长解码器

定长解码器不灵活,基本没有啥使用场景。

FixedLengthFrameDecoder:消息长度固定,累积读取到长度总和为定长 LEN 的报文后,就认为读取到了一个完整的消息,再将计数器置位,重新读取下一个数据报。

使用示例:

Service启动类配置:

image-20221105230933048

ServiceHandler类:

public class ServerHandler extends ChannelInboundHandlerAdapter {
    private int counter;

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 因为配置了StringDecoder 所以这里需要强转成String类型
        String body = (String) msg;
        System.out.println("服务端收到消息内容为:" + body + ", 收到消息次数:" + ++counter);

    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

ClientHandler类:

public class ClientHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {

        String strMsg = "fblinux.com";
        //连续发送
        for(int i=0; i< 100; i++){
            ctx.writeAndFlush(Unpooled.copiedBuffer(strMsg, CharsetUtil.UTF_8));
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {

        cause.printStackTrace();
        ctx.close();
    }
}

验证问题解决:

image-20221105231854554

5.解码器之行解码器

行解码器:使用换行符\n或者\r\n作为依据,遇到\n或者\r\n都认为是一条完整的消息。

行解码器提供两个构造函数:

LineBasedFrameDecoder(int maxLength)

LineBasedFrameDecoder(int maxLength,boolean stripDelimiter,boolean failFast)

对应的三个参数代表的意思:

maxLength: 表示一行最大的长度,如果超过这个长度依然没有检测到\n或者\r\n,抛出异常

stripDelimiter: 解码后的消息是否去除\n,\r\n分隔符,true:去除换行符,false:原样输出

failFast: 与maxLength联合使用,表示超过maxLength后,抛出TooLongFrameException的时机,true:超出maxLength后立即抛出异常,结束解码,false:等次消息包收完后,再抛出异常

使用示例:

Service启动类配置:

 image-20221105213241447

ServiceHandler类:

public class ServerHandler extends ChannelInboundHandlerAdapter {
    private int counter;

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 因为配置了StringDecoder 所以这里需要强转成String类型
        String body = (String) msg;
        System.out.println("服务端收到消息内容为:" + body + ", 收到消息次数:" + ++counter);

    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

ClientHandler类:

public class ClientHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {

        // 因为是按行分割,所以这里需要添加换行符
        String strMsg = "fblinux.com"+"\r\n";
        //连续发送
        for(int i=0; i< 100; i++){
            ctx.writeAndFlush(Unpooled.copiedBuffer(strMsg, CharsetUtil.UTF_8));
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {

        cause.printStackTrace();
        ctx.close();
    }
}

验证问题解决:

image-20221105233053091

6.解码器之分隔符解码器

DelimiterBasedFrameDecoder与LineBasedFrameDecoder类似,只不过更加通用,允许我们指定任意特殊字符作为分隔符。

分隔符解码器可以同时指定多个分隔符,如果指定多个分隔符,则会选择内容最短的一个分隔符作为依据

分隔符解码器有多个构造函数,区别在于以下四个参数的组合:

maxFrameLength: 表示一行最大的长度,如果超过这个长度依然没有检测自定义分隔符,将会抛出TooLongFrameException

stripDelimiter: 解码后的消息是否去除分隔符,true:去除换行符,false:原样输出;

failFast: 与maxFrameLength联合使用,表示超过maxFrameLength后,抛出异常的时机,true:超出maxLength后立即抛出异常,结束解码,false:等消息包收完后,再抛出异常;

delimiters: 分隔符,这个分隔符的类型要以ByteBuf的形式出现。

使用示例:

Service启动类配置:

image-20221105221657109

ClientHandler:

public class ClientHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {

        String message = "Netty is a NIO client server framework which enables quick&_" +
                "and easy development of network applications&_ " +
                "such as protocol servers and clients.&_" +
                " It greatly simplifies and streamlines&_" +
                "network programming such as TCP and UDP socket server.&_";

        ByteBuf mes = null;
        mes = Unpooled.buffer(message.getBytes().length);
        mes.writeBytes(message.getBytes());
        ctx.writeAndFlush(mes);

    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

测试验证:

image-20221105215419806

7.解码器之自定义长度解码器

自定义长度解码器,通过在消息头中定义消息长度字段来标志消息体的长度,然后根据消息的总长度来读取消息

自定义长度解码器能够适应各种复杂的通信协议格式

该解码器是四个解码器中相对来说稍显复杂一点,不是实现复杂,是因为以下两个方面的原因:

一是参数较多,二是准确计算各参数的值。

自定义长度解码器有多个构造函数,这些构造函数是7个参数的不同组合:

maxFrameLength: 消息的最大长度,超过最大长度抛出异常;

lengthFieldOffset: 长度域偏移量,指的是长度域位于整个数据包字节数组中的下标;

lengthFieldLength: 长度域自己的字节数长度;

lengthAdjustment: 默认值0,长度域的偏移量矫正,长度域的长度值是否要减去其它长度才是内容的值;

initialBytesToStrip: 默认值0,丢弃的起始字节数;

failFast: 默认值true,超过消息最大长度是否立即抛出异常;

ByteOrder: 字节序列模式,默认是大端模式,也可以设置为小端模式。

典型数据通信协议格式:

image-20221107221717592

图解各参数:

image-20221107221842318

使用示例:

ClientHandler:

public class ClientHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        for (int i = 0; i < 100; i++) {
            //发送自定义协议格式的消息
            short header=0x5A5A;
            byte[] msgtype=new byte[]{(byte)0x01};  // 构造器要求需要传入字节数组,所以这里定义成字节数组
            String strContent="fblinux.com";
            int contentLen=strContent.getBytes().length;

            //写入通道
            ctx.write(Unpooled.copyShort(header));
            ctx.write(Unpooled.copiedBuffer(msgtype));
            ctx.write(Unpooled.copyInt(contentLen));
            ctx.write(Unpooled.copiedBuffer(strContent, CharsetUtil.UTF_8));

            //发送消息
            ctx.flush();
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

ServiceHandler:

public class ServerHandler extends ChannelInboundHandlerAdapter {

    private int counter;

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

        //读取消息,并按照自定义的协议格式进行消息的处理
        ByteBuf in=(ByteBuf)msg;

        //打印报文
        String srcInfo= ByteBufUtil.hexDump(in).toUpperCase();
//        System.out.println("收到的原始报文: " + srcInfo);

        //进行报文的解析
        int header1=in.readByte();
        int msgtype1=in.readUnsignedShort();
        int contentLen1=in.readInt();
        ByteBuf bufContent1=in.readBytes(contentLen1);
        String strContent1=bufContent1.toString(CharsetUtil.UTF_8);
        System.out.println("服务端收到消息内容为:" + strContent1 + ", 收到消息次数:" + ++counter);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

服务端启动类:

image-20221106164703772

然后就可以启动程序,验证TCP粘包和拆包问题是否正常解决

转载请注明:西门飞冰的博客 » Netty 四个粘包拆包解码器

喜欢 (1)or分享 (0)