Netty入站出站介绍及借助使用到责任链简单实现实现网络编程中拆包和粘包

ゞ浴缸里的玫瑰 2023-02-17 14:30 182阅读 0赞

### Netty入站出站事件 ###

**入站事件：**  
通常指I/O线程生成入站数据。（通俗理解：从socket底层往上冒的事件都是入站）比如EventLoop收到selector的OP\_READ事件，入站处理器调用socketChannel.read(ByteBuffer)接收到数据后，这将通道ChannelPipeline中包含的下一个channelRead方法被调用。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhamluZ19mZWl5dWU_size_16_color_FFFFFF_t_70]

**出站事件：**  
经常是指I/O线程执行实际的输出操作。（通俗理解：主动往socket底层操作的事件都是出站）比如bind方法的用意是请求server socket绑定的SocketAddress,这将导致通道的ChannelPipline包含下一个出站处理器中bind方法被调用。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhamluZ19mZWl5dWU_size_16_color_FFFFFF_t_70 1]

### Netty通过责任链实现出入站事件处理 ###

我们在使用责任链的时候更多的是  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhamluZ19mZWl5dWU_size_16_color_FFFFFF_t_70 2]  
netty的责任链，与我们我们经常使用到责任链相比多了ChannelPipeline来维护这个责任链。  
ChannelPipeline链中存储的也并不是直接的处理的处理器，而是包裹handler处理器的DefaultChannelHandlerContext。  
对于我这边理解之所以引入ChannelPipeline这个上下文，是因为在在使用handler处理器的时候，需要有添加和删除handler的操作，用ChannelPipeline这个这个责任链能够更好管理hanler处理。

DefaultChannelHandlerContext这种包装的代替handler的原因，是在handler处理的时候，可能会经常遇到另外线程池的处理，而更好关注业务本身用DefaultChannelHandlerContext代替了handler.  
![在这里插入图片描述][20200614172306372.png]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhamluZ19mZWl5dWU_size_16_color_FFFFFF_t_70 3]

那我之所以需要分入站与出站，这个是因为我们在添加责任链的时候，入站事件只会处理入站事件，而出站同样也只能处理出站事件。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhamluZ19mZWl5dWU_size_16_color_FFFFFF_t_70 4]

### 借助责任链简单解决一下网络编程中拆包和粘包 ###

**拆包和粘包出现的原因：**  
（1）由于Nagle算法可能会将一些，要发送的数据小于TCP发送缓冲区的大小，TCP将多次写入缓冲区的数据一次发送出去，将会发生粘包；

（2）接收数据端的应用层没有及时读取接收缓冲区中的数据，将发生粘包；

（3）要发送的数据大于TCP发送缓冲区剩余空间大小，将会发生拆包；

（4）待发送数据大于MSS（最大报文长度），TCP在传输前将进行拆包。即TCP报文长度-TCP头部长度>MSS。

**拆包和粘包出现的解决方案：**  
由于底层的TCP无法理解上层的业务数据，所以在底层是无法保证数据包不被拆分和重组的，这个问题只能通过上层的应用协议栈设计来解决，根据业界的主流协议的解决方案，归纳如下：

（1）、使用带消息头的协议、消息头存储消息开始标识及消息长度信息，服务端获取消息头的时候解析出消息长度，然后向后读取该长度的内容。

（2）、设置定长消息，服务端每次读取既定长度的内容作为一条完整消息。

（3）、设置消息边界，服务端从网络流中按消息编辑分离出消息内容。

**Netty自带的拆包器:**  
（1）、固定长度的拆包器FixedLengthFrameDecoder:如果应用层协议非常简单,每个数据包的长度都是固定的,需要把拆包器加到Pipeline,Netty把指定长度的数据包(ByteBuf)传递到下一个ChannelHandler.  
（2）、行拆包器LineBasedFrameDecoder:发送端发送数据包的时候,每个数据包之间以换行符作为分隔,接收端通过LineBasedFrameDecoder将粘包的ByteBuf拆分成一个个完整的应用层数据包.  
（3）、分隔符拆包器DelimiterBasedFrameDecoder:行拆包器的通用版本,自定义分隔符.  
（4）、基于长度域拆包器LengthFieldBasedFrameDecoder:最通用的一种拆包器,只要自定义协议包含长度域字段均使用此拆包器实现应用层拆包。使用LengthFieldBasedFrameDecoder需要长度域相对整个数据包的偏移量和长度域的长度构造拆包器,Pipeline最前面添加此拆包器。

**代码编写：**  
1、这里通过最简单 固定发送长度这种最简单的粗略的方案来演示这个问题与处理。这里采用每次发送210个字节。

服务端代码：

public class XNettyServer {
        public static void main(String[] args) throws Exception {
            // 1、 线程定义
            // accept 处理连接的线程池线程大小为默认当前cpu的两倍
            //DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt("io.netty.eventLoopThreads", Runtime.getRuntime().availableProcessors() * 2));
            EventLoopGroup acceptGroup = new NioEventLoopGroup();
            // read io 处理数据的线程池
            EventLoopGroup readGroup = new NioEventLoopGroup();
            //业务线程，防止处理线程的任务因为任务阻塞
            NioEventLoopGroup bizGroup = new NioEventLoopGroup(4,new DefaultThreadFactory("biz-thread-pool-"));
            try {
                ServerBootstrap b = new ServerBootstrap();
                b.group(acceptGroup, readGroup);
                // 2、 选择TCP协议，NIO的实现方式
                b.channel(NioServerSocketChannel.class);
               // ChannelOption.SO_KEEPALIVE表示是否开启TCP底层心跳机制,true为开启
                 b.childOption(ChannelOption.SO_KEEPALIVE, true);
                 //ChannelOption.SO_REUSEADDR表示端口释放后立即就可以被再次使用,因为一般来说,一个端口释放后会等待两分钟之后才能再被使用
                 b.childOption(ChannelOption.SO_REUSEADDR, true);
                 // ChannelOption.TCP_NODELAY表示是否开始Nagle算法,true表示关闭,false表示开启,通俗地说,如果要求高实时性,有数据发送时就马上发送,就关闭,如果需要减少发送次数减少网络交互就开启
                b.childOption(ChannelOption.TCP_NODELAY, false);
                //设置发送缓冲区大小
              //  b.childOption(ChannelOption.SO_SNDBUF, 32 * 1024);
                //设置接收缓冲区大小
              //  b.childOption(ChannelOption.SO_RCVBUF, 32 * 1024);
                XHandller xHandller = new XHandller();
                b.childHandler(new ChannelInitializer<SocketChannel>() {
    
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        // 3、 职责链定义（请求收到后怎么处理）
                        ChannelPipeline pipeline = ch.pipeline();
    
                        pipeline.addLast(new XDecoder());
    
                        pipeline.addLast(bizGroup,xHandller);
                    }
                });
                // 4、 绑定端口
                System.out.println("启动成功，端口 9999");
                b.bind(9999).sync().channel().closeFuture().sync();
            } finally {
                acceptGroup.shutdownGracefully();
                readGroup.shutdownGracefully();
            }
        }
    }

根据定义每次发送210个字节来进行解码

public class XDecoder extends ByteToMessageDecoder {
        static final int PACKET_SIZE = 220;
    
        // 用来临时保留没有处理过的请求报文
        ByteBuf tempMsg = Unpooled.buffer();
    
        @Override
        protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
            System.out.println("收到了一次数据包，长度是：" + in.readableBytes());
            // in 请求的数据
            // out 将粘在一起的报文拆分后的结果保留起来
    
            // 1、 合并报文
            ByteBuf message = null;
            int tmpMsgSize = tempMsg.readableBytes();
            // 如果暂存有上一次余下的请求报文，则合并
            if (tmpMsgSize > 0) {
                message = Unpooled.buffer();
                message.writeBytes(tempMsg);
                message.writeBytes(in);
                System.out.println("合并：上一数据包余下的长度为：" + tmpMsgSize + ",合并后长度为:" + message.readableBytes());
            } else {
                message = in;
            }
    
            // 2、 拆分报文
       
            int size = message.readableBytes();
            int counter = size / PACKET_SIZE;
            for (int i = 0; i < counter; i++) {
                byte[] request = new byte[PACKET_SIZE];
                // 每次从总的消息中读取210个字节的数据
                message.readBytes(request);
    
                // 将拆分后的结果放入out列表中，交由后面的业务逻辑去处理
                out.add(Unpooled.copiedBuffer(request));
            }
    
            // 3、多余的报文存起来
            // 第一个报文： i+  暂存
            // 第二个报文： 1 与第一次
            size = message.readableBytes();
            if (size != 0) {
                System.out.println("多余的数据长度：" + size);
                // 剩下来的数据放到tempMsg暂存
                tempMsg.clear();
                tempMsg.writeBytes(message.readBytes(size));
            }
    
        }
    
    }

业务处理器：

/**
     * 后续处理handdler
     */
    @ChannelHandler.Sharable
    public class XHandller extends ChannelInboundHandlerAdapter {
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            ctx.flush();
        }
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            // 输出 bytebuf
            ByteBuf buf = (ByteBuf) msg;
            byte[] content = new byte[buf.readableBytes()];
            buf.readBytes(content);
    
            System.out.println(Thread.currentThread().getName() + ":" + new String(content));
            //释放堆外内存 重复利用，gc释放
           // buf.release(); //也快交流后面处理缓冲区的释放
            ctx.fireChannelRead(msg);
        }
    
        // 异常
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    
    }

客户端：

public class TonySocketClient {
        public static void main(String[] args) throws Exception {
            Socket socket = new Socket("localhost", 9999);
            OutputStream outputStream = socket.getOutputStream();
    
            // 消息长度固定为 220字节，包含有
            // 2. 消息内容字符串长度最多70。 按一个汉字3字节，内容的最大长度为210字节
            byte[] request = new byte[220];
            byte[] userId = "useridtest".getBytes();
            byte[] content = "发送内容test23444444444444444444444444445test".getBytes();
            System.arraycopy(userId, 0, request, 0, 10);
            System.arraycopy(content, 0, request, 10, content.length);
    		//通过并发方式来展示发送消息，这样更容易出现粘包
            CountDownLatch countDownLatch = new CountDownLatch(10);
            for (int i = 0; i < 10; i++) {
                new Thread(() -> {
                    try {
                        outputStream.write(request);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                    countDownLatch.countDown();
                }).start();
            }
    
            countDownLatch.await();
            Thread.sleep(2000L); // 两秒后退出
            socket.close();
        }
    }

处理结果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhamluZ19mZWl5dWU_size_16_color_FFFFFF_t_70 5]  
当我注释掉添加解吗处理器：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhamluZ19mZWl5dWU_size_16_color_FFFFFF_t_70 6]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhamluZ19mZWl5dWU_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200614162415674.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhamluZ19mZWl5dWU=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhamluZ19mZWl5dWU_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200614162752226.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhamluZ19mZWl5dWU=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhamluZ19mZWl5dWU_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200614171600990.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhamluZ19mZWl5dWU=,size_16,color_FFFFFF,t_70
[20200614172306372.png]: https://img-blog.csdnimg.cn/20200614172306372.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhamluZ19mZWl5dWU_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20200614171306591.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhamluZ19mZWl5dWU=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhamluZ19mZWl5dWU_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20200614173715815.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhamluZ19mZWl5dWU=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhamluZ19mZWl5dWU_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/20200614182454666.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhamluZ19mZWl5dWU=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhamluZ19mZWl5dWU_size_16_color_FFFFFF_t_70 6]: https://img-blog.csdnimg.cn/20200614183003935.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhamluZ19mZWl5dWU=,size_16,color_FFFFFF,t_70