ICode9
精准搜索请尝试： 精确搜索
首页
编程语言>
数据库
系统相关
互联网
其他分享
Python
Java
JavaScript
android
PHP
首页 > 其他分享> 文章详细

Netty网络框架学习笔记-11(TCP 粘包和拆包_2022.06.08)

2022-06-20 18:02:18  阅读：117  来源： 互联网

标签：11 Netty 读取 int 08 ctx LENGTH new public


Netty网络框架学习笔记-11(TCP 粘包和拆包_2022.06.08)

TCP 粘包和拆包基本介绍

TCP 是面向连接的，面向字节流的，提供高可靠性服务。收发两端（客户端和服务器端）都要有一一成对的 socket，

因此，发送端为了将多个发给接收端的包，更有效的发给对方，使用了优化方法（Nagle 算法），将多次间隔 较小且数据量小的数据，合并成一个大的数据块(缓冲区)，然后进行封包。这样做虽然提高了效率，但是接收端就难于分辨出完整的数据包了，因为面向流的通信是无消息保护边界的

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

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

为什么UDP没有粘包

粘包拆包问题在数据链路层、网络层以及传输层都有可能发生。日常的网络应用开发大都在传输层进行，由于UDP有消息保护边界，不会发生粘包拆包问题，因此粘包拆包问题只发生在TCP协议中。

由于 TCP 无消息保护边界, 需要在接收端处理消息边界问题，也就是我们所说的粘包、拆包问题, 如下图:

XDKexO.png

常见的解决方案

常见的有四种:

  • 发送端将每个包都封装成固定的长度，比如100字节大小。如果不足100字节可通过补0或空等进行填充到指定长度
  • 发送端在每个包的末尾使用固定的分隔符，例如\r\n。如果发生拆包需等待多个包发送过来之后再找到其中的\r\n进行合并；例如，FTP协议
  • 将消息分为头部和消息体，头部中保存整个消息的长度，只有读取到足够长度的消息之后才算是读到了一个完整的消息
  • 通过自定义协议进行粘包和拆包的处理

Netty的粘包

在编写 Netty 程序时，如果没有做处理，就会发生粘包和拆包的问题

演示:

  • 服务端( 只演示处理器类不同)

服务端的引导配置类都和前面的一样(这里省略引导类), 只是自定义处理器不一样。

@Slf4j
public class NettyTcpServiceHandle extends SimpleChannelInboundHandler<String> {

    private static int count = 0;

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        log.info("HeartbeatServerHandle-读取到信息:{}", msg);
        log.info("HeartbeatServerHandle-读取到多少次信息:{}", ++count);
        for (int i = 0; i < 5; i++) {
            ctx.writeAndFlush(UUID.fastUUID().toString());
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.channel().close();
    }
}
  • 客户端( 只演示处理器类不同)
@Slf4j
public class NettyTcpClientHandle extends SimpleChannelInboundHandler<String> {

    private static int count = 0;

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        log.info("NettyTcpClientHandle-读取到信息:{}", msg);
        log.info("NettyTcpClientHandle-读取到多少次信息:{}", ++count);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        for (int i = 0; i < 5; i++) {
            ctx.writeAndFlush(UUID.fastUUID().toString());
        }
    }
    
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.channel().close();
    }
}
  • 结果:
14:13:26.139 [nioEventLoopGroup-3-1] INFO com.zhihao.netty.tcp.NettyTcpServiceHandle - HeartbeatServerHandle-读取到信息:3706f07d-d07d-4c1c-acb3-126c01a74da0b6361489-8ed7-4579-821d-d8b152dbbf6917ec1195-06d6-4eef-bb5f-a9e8a0058297acb2ea14-82b6-42cd-bfa5-4f5ca4d73974044d22e4-ce5e-439d-8247-8ebfcf39649b
14:13:26.139 [nioEventLoopGroup-3-1] INFO com.zhihao.netty.tcp.NettyTcpServiceHandle - HeartbeatServerHandle-读取到多少次信息:1
// ------------------------------------------------------
14:13:26.155 [nioEventLoopGroup-2-1] INFO com.zhihao.netty.tcp.NettyTcpClientHandle - NettyTcpClientHandle-读取到信息:46d05f67-6b61-42d3-aacd-1a6535584a9c4797c824-8739-4315-9813-9b8e8dd0ac1198937163-1194-48b1-bfe4-6a40fa0169fe5cd9e69a-a6d5-4965-8f26-42f6dcb0dd64780751e3-567a-439f-882e-1c227bb14fcc
14:13:26.155 [nioEventLoopGroup-2-1] INFO com.zhihao.netty.tcp.NettyTcpClientHandle - NettyTcpClientHandle-读取到多少次信息:1

可以看出, 客户端 , 循环了10次发送消息, 服务端是按照一次接收一个数据包解决, 这就出现了粘包问题, 在没有配置的情况下, 接收方也不知道如何处理。

Netty对粘包和拆包问题的处理

Netty对解决粘包和拆包的方案做了抽象，提供了一些解码器（Decoder）来解决粘包和拆包的问题。如：

  • LineBasedFrameDecoder：以行为单位进行数据包的解码；
  • DelimiterBasedFrameDecoder：以特殊的符号作为分隔来进行数据包的解码；
  • FixedLengthFrameDecoder：以固定长度进行数据包的解码；
  • LenghtFieldBasedFrameDecode：适用于消息头包含消息长度的协议（最常用）；

基于Netty进行网络读写的程序，可以直接使用这些Decoder来完成数据包的解码。对于高并发、大流量的系统来说，每个数据包都不应该传输多余的数据（所以补齐的方式不可取），LenghtFieldBasedFrameDecode更适合这样的场景。

使用自定义协议 + 编解码器 来解决

这里使用LenghtFieldBasedFrameDecode 解码器

1.0 自定义协议消息体

@Data
@AllArgsConstructor
@NoArgsConstructor
public class MessageProtocol {

    /**
     * 消息长度
     */
    private Long messageLength;

    /**
     * 内容
     */
    private String content;
}

2.0 自定义协议消息体解码器

public class MyLenghtFieldBasedFrameDecode extends LengthFieldBasedFrameDecoder {

    /**
     * @param maxFrameLength      帧的最大长度
     * @param lengthFieldOffset   length字段偏移的地址 (长度字段所在索引)
     * @param lengthFieldLength   length字段所占的字节长
     * @param lengthAdjustment    修改帧数据长度字段中定义的值，可以为负数 因为有时候我们习惯把头部记入长度,若为负数,则说明要推后多少个字段
     * @param initialBytesToStrip 解析时候跳过多少个长度
     * @param failFast            为true，当frame长度超过maxFrameLength时立即报TooLongFrameException异常，为false，读取完整个帧再报异
     */
    public MyLenghtFieldBasedFrameDecode(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength, int lengthAdjustment, int initialBytesToStrip, boolean failFast) {
        super(maxFrameLength, lengthFieldOffset, lengthFieldLength, lengthAdjustment, initialBytesToStrip, failFast);
    }

    @Override
    protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
        // 先调用父类解码, 按照构造方法参数进行解码
        in = (ByteBuf) super.decode(ctx, in);
        if (in == null) {
            return null;
        }

        if (in.readableBytes() <= 0) {
            throw new RuntimeException("字节数不足");
        }
        //读取messageLength字段,  字段类型是什么, 就读取什么类型, 如果同一个实体有多个相同字段则是按顺序读取
        long length = in.readLong();

        if (in.readableBytes() != length) {
            throw new RuntimeException(in.readableBytes() + ":>>>标记的长度不符合实际长度:>>>" + length);
        }
        //读取body
        byte[] bytes = new byte[in.readableBytes()];
        in.readBytes(bytes);
        // PS: 如果后面后续有转换成为其他实体的的处理器的情况下, 这里应返回ByteBuf类型
        return new MessageProtocol(length, new String(bytes, StandardCharsets.UTF_8));
    }
}

3.0 自定义协议消息体编码器

@Slf4j
public class MyMessageProtocolEncoder extends MessageToByteEncoder<MessageProtocol> {

    @Override
    protected void encode(ChannelHandlerContext ctx, MessageProtocol msg, ByteBuf out) throws Exception {
        Long messageLength = Optional.ofNullable(msg).map(MessageProtocol::getMessageLength).orElse(0L);
        if (messageLength.longValue() == 0L) {
            return;
        }
        if (log.isDebugEnabled()) {
            log.debug("MyMessageProtocolEncoder-msg:{}", msg);
        }
        out.writeLong(messageLength);
        out.writeBytes(msg.getContent().getBytes(StandardCharsets.UTF_8));
    }
}

4.0 服务端使用

@Slf4j
public class NettySolveTcpService {

    private static final int MAX_FRAME_LENGTH = 1024 * 1024;  //最大长度
    private static final int LENGTH_FIELD_OFFSET = 0;  //长度偏移  这个指的是你的长度字段在消息整包中的开始位置, 实体第一个字段就是长度
    private static final int LENGTH_FIELD_LENGTH = 8;  //长度字段所占的字节数  long=8字节
    private static final int LENGTH_ADJUSTMENT = 0;
    private static final int INITIAL_BYTES_TO_STRIP = 0;

    public static void main(String[] args) {
        NioEventLoopGroup bossGroup = new NioEventLoopGroup(2);
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();

        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap = bootstrap.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .childOption(ChannelOption.SO_BACKLOG, 128)
                .childOption(ChannelOption.SO_KEEPALIVE, true)
                .handler(new LoggingHandler(LogLevel.INFO)) // boosGroup服务器处理程序, 日志
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        // 添加自定义实现编解码处理器
                        pipeline.addLast(new MyMessageProtocolEncoder());
                        pipeline.addLast(new MyLenghtFieldBasedFrameDecode(MAX_FRAME_LENGTH, LENGTH_FIELD_OFFSET,
                                LENGTH_FIELD_LENGTH, LENGTH_ADJUSTMENT, INITIAL_BYTES_TO_STRIP, Boolean.FALSE));
                        // 后续还可以增加接收上面编解码处理结果的编解码处理器转换成为我们想要的对象
                        // 添加自己的处理器
                        pipeline.addLast(new NettySolveTcpServiceHandle());
                    }
                });

        try {
            ChannelFuture channelFuture = bootstrap.bind(new InetSocketAddress("127.0.0.1", 8888)).sync();
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

4.0.1 服务端处理器

/**
 * @Author: ZhiHao
 * @Date: 2022/4/24 17:44
 * @Description: 这里接收的就是 消息协议实体了
 * @Versions 1.0
 **/
@Slf4j
public class NettySolveTcpServiceHandle extends SimpleChannelInboundHandler<MessageProtocol> {

    private static int count = 0;

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, MessageProtocol msg) throws Exception {
        log.info("HeartbeatServerHandle-读取到信息:{}", msg);
        log.info("HeartbeatServerHandle-读取到多少次信息:{}", ++count);
        for (int i = 0; i < 5; i++) {
            String str = UUID.fastUUID() + "||";
            ctx.writeAndFlush(new MessageProtocol(Long.valueOf(str.length()), str));
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.info("HeartbeatServerHandle-通道关闭, 异常信息:", cause);
        ctx.channel().close();
    }
}

5.0 客户端使用

@Slf4j
public class NettySolveTcpClient {

    private static final int MAX_FRAME_LENGTH = 1024 * 1024;  //最大长度
    private static final int LENGTH_FIELD_OFFSET = 0;  //这个指的是你的长度字段在消息整包中的开始位置, 实体第一个字段就是长度
    private static final int LENGTH_FIELD_LENGTH = 8;  //长度字段所占的字节数  long=8字节
    private static final int LENGTH_ADJUSTMENT = 0;
    private static final int INITIAL_BYTES_TO_STRIP = 0;

    public static void main(String[] args) {
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();

        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(workerGroup)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        // 添加自定义实现编解码处理器
                        pipeline.addLast(new MyMessageProtocolEncoder());
                        pipeline.addLast(new MyLenghtFieldBasedFrameDecode(MAX_FRAME_LENGTH, LENGTH_FIELD_OFFSET,
                                LENGTH_FIELD_LENGTH, LENGTH_ADJUSTMENT, INITIAL_BYTES_TO_STRIP, Boolean.FALSE));
                        // 后续还可以增加接收上面编解码处理结果的编解码处理器转换成为我们想要的对象
                        // 添加自己的处理器
                        pipeline.addLast(new NettySolveTcpClientHandle());
                    }
                });

        try {
            ChannelFuture channelFuture = bootstrap.connect(new InetSocketAddress("127.0.0.1", 8888)).sync();

            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            workerGroup.shutdownGracefully();
        }
    }
}

5.0.1 客户端处理器

@Slf4j
public class NettySolveTcpClientHandle extends SimpleChannelInboundHandler<MessageProtocol> {

    private static int count = 0;

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, MessageProtocol msg) throws Exception {
        log.info("NettyTcpClientHandle-读取到信息:{}", msg);
        log.info("NettyTcpClientHandle-读取到多少次信息:{}", ++count);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        for (int i = 0; i < 5; i++) {
            String str = UUID.fastUUID() + "||";
            ctx.writeAndFlush(new MessageProtocol(Long.valueOf(str.length()), str));
            log.info("NettyTcpClientHandle-{}",i);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.info("NettyTcpClientHandle-通道关闭, 异常信息:", cause);
        ctx.channel().close();
    }
}

结果

17:14:38.070 [nioEventLoopGroup-3-1] INFO com.zhihao.netty.tcp.solveTcp.NettySolveTcpServiceHandle - HeartbeatServerHandle-读取到信息:MessageProtocol(messageLength=38, content=2621667d-125d-4675-b76e-1536ab48f83d||)
17:14:38.070 [nioEventLoopGroup-3-1] INFO com.zhihao.netty.tcp.solveTcp.NettySolveTcpServiceHandle - HeartbeatServerHandle-读取到多少次信息:1
17:14:38.070 [nioEventLoopGroup-3-1] INFO com.zhihao.netty.tcp.solveTcp.NettySolveTcpServiceHandle - HeartbeatServerHandle-读取到信息:MessageProtocol(messageLength=38, content=fa26ed5a-dfc0-490a-8e4e-e4bc7ebe0252||)
17:14:38.071 [nioEventLoopGroup-3-1] INFO com.zhihao.netty.tcp.solveTcp.NettySolveTcpServiceHandle - HeartbeatServerHandle-读取到多少次信息:2
17:14:38.071 [nioEventLoopGroup-3-1] INFO com.zhihao.netty.tcp.solveTcp.NettySolveTcpServiceHandle - HeartbeatServerHandle-读取到信息:MessageProtocol(messageLength=38, content=9441a190-8c80-4015-b41c-2f4a8285c059||)
17:14:38.071 [nioEventLoopGroup-3-1] INFO com.zhihao.netty.tcp.solveTcp.NettySolveTcpServiceHandle - HeartbeatServerHandle-读取到多少次信息:3
17:14:38.071 [nioEventLoopGroup-3-1] INFO com.zhihao.netty.tcp.solveTcp.NettySolveTcpServiceHandle - HeartbeatServerHandle-读取到信息:MessageProtocol(messageLength=38, content=d8f080fc-9706-4118-a08b-3a67f8cd3250||)
17:14:38.071 [nioEventLoopGroup-3-1] INFO com.zhihao.netty.tcp.solveTcp.NettySolveTcpServiceHandle - HeartbeatServerHandle-读取到多少次信息:4
17:14:38.071 [nioEventLoopGroup-3-1] INFO com.zhihao.netty.tcp.solveTcp.NettySolveTcpServiceHandle - HeartbeatServerHandle-读取到信息:MessageProtocol(messageLength=38, content=9f94a873-5a01-4a6f-bc5b-9d20ac3305a1||)
17:14:38.071 [nioEventLoopGroup-3-1] INFO com.zhihao.netty.tcp.solveTcp.NettySolveTcpServiceHandle - HeartbeatServerHandle-读取到多少次信息:5
// ----------------------------客户端日志省略

从上面结果来看, 已经解决了粘包问题。 拆包也是同样使用这种方式处理。

扩展:

LengthFieldBasedFrameDecoder的使用解析

1

标签：11,Netty,读取,int,08,ctx,LENGTH,new,public
来源： https://www.cnblogs.com/zhihao-plus/p/16394158.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

关于我们 | 联系我们 | 留言反馈

专注分享技术，共同学习，共同进步。侵权联系[81616952@qq.com]

Copyright (C)ICode9.com, All Rights Reserved.

ICode9版权所有