Netty核心组件 + 消息边界问题的解决

痛定思痛。 2022-10-11 13:46 209阅读 0赞

**推荐先阅读关于NIO的文章作为预备知识，传送地址：**

**[NIO三大组件介绍及ByteBuffer的使用][NIO_ByteBuffer]**

**[NIO多路复用之Selector的使用][NIO_Selector]**

# Netty核心组件 #

### 文章目录 ###

*  Netty核心组件
 *   *  一、入门案例
     *   *  1. 概述
         *  2. 入门案例
         *  3. 流程梳理
     *  二、组件
     *   *  1. EventLoop
         *  2. Channel
         *   *  2.1 ChannelFuture
             *  2.2 CloseFuture
             *  2.3 Netty为什么使用异步
         *  3. Future & Promise
         *   *  3.1 JDK-Future
             *  3.2 Netty-Future
             *  3.3 Netty-Promise
         *  4. Handler & Pipeline
         *  5. ByteBuf
         *   *  5.1 零拷贝
             *   *  5.1.1 slice
                 *  5.1.2 duplicate
                 *  5.1.3 CompositeByteBuf
                 *  5.1.4 Unpooled
                 *  5.1.5 copy
             *  5.2 ByteBuf优势
     *  三、消息边界问题
     *   *  1. 现象分析
         *  2. 解决方案

## 一、入门案例 ##

### 1. 概述 ###

Netty 是一个异步的、基于事件驱动（selector的多路复用机制）的网络应用框架，用于快速开发可维护、高性能的网络服务器和客户端。

### 2. 入门案例 ###

开发一个简单的服务器端和客户端

*  客户端向服务器端发送 hello, world
 *  服务器仅接收，不返回

**加入依赖：**

<dependency>
        <groupId>io.netty</groupId>
        <artifactId>netty-all</artifactId>
        <version>4.1.39.Final</version>
    </dependency>

**服务器端：**

public class HelloServer { 
        public static void main(String[] args) { 
    
            // 1. 启动器，负责组装 netty 组件，启动服务器
            new ServerBootstrap()
    
                // 2. 类似于之前提到的多线程优化处理不同事件的selector，EventLoop=selector+thread, group组表示有多个EventLoop
                .group(new NioEventLoopGroup())
    
                // 3. 服务器的ServerSocketChannel实现，此处选择的是NIO对ServerSocketChannel的封装
                .channel(NioServerSocketChannel.class)
    
                // 4. child类似于多线程优化中提到的EventLoop，负责读写处理，childHandler表示worker应该做的操作（handler）
                .childHandler(
    
                // 5. Initializer初始化器，在initChannel方法中添加handler，定义worker要做的操作
                // ChannelInitializer的initChannel方法会在连接建立之后自动调用
                // NioSocketChannel表示连接建立成功之后的数据传输通道
                new ChannelInitializer<NioSocketChannel>() { 
    
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception { 
    
                        // 6. 在流水线pipeline中添加具体的handler，即worker应该做的操作
                        ch.pipeline().addLast(new StringDecoder()); // 解码，将ByteBuf（字节）转换为字符串
    
                        ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {  // 自定义的业务处理器
    
                            // 关注读事件，表示在读事件发生之后接收消息，参数msg表示接收到的解码后的消息
                            @Override
                            public void channelRead(ChannelHandlerContext ctx,Object msg) throws Exception { 
                                System.out.println(msg); // 打印上一步解码后的字符串
                            }
                        });
                    }
                })
    
                // 7. 绑定监听端口
                .bind(8080);
        }
    }

**代码解读：**

*  第 2 步，创建 `NioEventLoopGroup` ，可以简单理解为 `线程池 + Selector` ，后面会详细展开
 *  第 3 步，定义服务器的 Channel 实现类，其中 `NioServerSocketChannel` 表示基于 NIO 的服务器端实现，其它实现还有：
    
    ![image-20210628135710527][]
 *  第 4 步 `childHandler`，表示接下来添加的处理器都是给 SocketChannel 用的，而不是给 ServerSocketChannel 用的
    
     *  ChannelInitializer 处理器（仅执行一次），它的作用是待客户端 SocketChannel 建立连接后，执行 `initChannel` 以便添加更多的处理器
 *  第 6 步的 `StringDecoder`，表示 SocketChannel 的处理器，进行解码操作，即 `ByteBuf => String`

**客户端代码：**

public class HelloClient { 
        public static void main(String[] args) throws InterruptedException { 
            // 1. 启动器，与服务器创建的对象不一致
            new Bootstrap()
                // 2. 添加EventLoop
                .group(new NioEventLoopGroup())
                // 3. 客户端的channel实现
                .channel(NioSocketChannel.class)
                // 4. 添加处理器
                .handler(new ChannelInitializer<NioSocketChannel>() { 
                    //在连接建立后被调用
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception { 
                        ch.pipeline().addLast(new StringEncoder()); //编码器，与服务器对称
                    }
                })
                // 5. 连接到服务器
                .connect(new InetSocketAddress("localhost", 8080))
                .sync()
                .channel()
                // 6. 向服务器发送数据
                .writeAndFlush("hello, world");
        }
    }

**代码解读：**

*  第 4 步，添加 SocketChannel 的处理器，ChannelInitializer 处理器（仅执行一次），它的作用是待客户端 SocketChannel 建立连接后，执行 `initChannel` 方法以便添加更多的处理器
 *  第 5 步的三个方法
    
     *  指定要连接的服务器和端口
     *  Netty 中很多方法都是异步的，如 `connect()`，这时需要使用 `sync()` 方法等待 connect 建立连接完毕（阻塞直到连接建立）
     *  获取 channel 对象，即通道抽象，可以进行数据读写操作
 *  第 6 步，写入消息并清空缓冲区
 *  第 4 步的 `StringEncoder`，发送的消息会经过 handler 处理，这里是将 `String => ByteBuf`，即编码
 *  数据经过网络传输，到达服务器端，服务器端第 6 步的 StringDecoder、业务处理器先后被触发，完成一个流程

### 3. 流程梳理 ###

**数字表示执行的顺序：**

![image-20210625162800228][]

*  第2步中，一开始关心的是accept事件
 *  第4步中，服务器添加了处理器，但不会立即执行，连接建立以后才会执行 `initChannel` 方法
 *  执行完第6步，服务器已经启动成功，EventLoop 线程监听 accept 事件
 *  第10步中，客户端添加了一个处理器，但不会立即执行，连接建立以后才会执行 `initChannel` 方法
 *  第15步中，发送的数据要经过编码
    
     *  注意第15步并不是添加一个处理器，处理器的添加操作在第12步中

**一些概念的解释：**

*  把 channel 理解为数据的通道
 *  把处理器的参数 msg 理解为流动的数据，最开始输入的类型是 ByteBuf，但经过 pipeline（流水线）（多个handler连在一起） 的加工，会变成其它类型对象，最后输出又变成 ByteBuf
 *  把 handler 理解为数据的处理工序
    
     *  工序有多道，合在一起就是 pipeline，pipeline 负责发布事件，传播给每个 handler 来处理， handler 对自己感兴趣的事件进行处理（重写了相应事件处理方法）
     *  handler 分为 Inbound（入站） 和 Outbound（出站） 两类，后面再解释
 *  EventLoop 是真正用来处理数据的
    
     *  一个 EventLoop 可以管理多个 channel 的 IO 操作，并且一旦 EventLoop 负责了某个 channel，就要负责到底（绑定）
        
         *  目的是为了线程安全，多个 EventLoop 处理同一个channel会造成线程不安全问题
     *  EventLoop 既可以执行 IO 操作，也可以进行任务处理，每个 EventLoop 都有任务队列，队列里可以堆放多个 channel 的待处理任务，任务分为普通任务、定时任务
        
         *  EventLoop 底层使用线程池
     *  EventLoop 依次按照 pipeline 中 handler 的规划处理数据
        
         *  可以为每个 handler 指定不同的 EventLoop，也就是处理过程中切换了线程（EventLoop）

## 二、组件 ##

### 1. EventLoop ###

**EventLoop** 本质是一个单线程执行器（同时维护了一个 Selector），用来处理 Channel 上的 IO 事件。

一个 EventLoop 对应着一个线程。

它的继承关系比较复杂：

*  一方面是继承自 `j.u.c.ScheduledExecutorService`，因此包含了线程池中所有的方法
 *  另一方面是继承自 Netty 自己的 `OrderedEventExecutor`，保证了处理 IO 事件的顺序
    
     *  提供了 `boolean inEventLoop(Thread thread)` 方法判断一个线程是否属于此 EventLoop
     *  提供了 `parent()` 方法来判断自己属于哪个 EventLoopGroup

**EventLoopGroup** 是一组 EventLoop，Channel 一般会调用 EventLoopGroup 的 `register` 方法来绑定其中一个 EventLoop，后续这个 Channel 上的 IO 事件都由此 EventLoop 来处理（保证了 IO 事件处理时的线程安全）

*  继承自 Netty 自己的 EventExecutorGroup
    
     *  实现了 Iterable 接口，提供遍历 EventLoop 的能力
     *  另有 `next()` 方法获取集合中的下一个 EventLoop

**执行普通任务、定时任务：**

// 1. 创建EventLoopGroup
    EventLoopGroup group = new NioEventLoopGroup(2); // 可以处理三种任务：IO事件、普通任务、定时任务
    // 如果不指定线程数，则线程数为CPU核心数*2
    
    //EventLoopGroup group = new DefaultEventLoopGroup(); // 可以处理两种任务：普通任务、定时任务
    
    // 2. 获取下一个EventLoop（此时设定了两个）
    System.out.println(group.next()); //第一次调用，获取第一个EventLoop
    //io.netty.channel.nio.NioEventLoop@2ef5e5e3
    
    group.next(); //第二次调用
    System.out.println(group.next()); //第三次调用，重新获取第一个事件循环对象
    //io.netty.channel.nio.NioEventLoop@2ef5e5e3
    
    // 3. 执行普通任务
    // EventLoop拥有如execute、submit等从线程池中继承下来的方法
    group.next().execute(() -> {  //将任务提交给EventLoopGroup中某一个EventLoop（线程）执行
    
        System.out.println("ok"); //此代码不是由主线程执行
    });
    // 执行普通任务的目的是，可以对任务进行异步处理，某些耗时的工作可以不由当前线程完成，交由事件循环组中某一线程完成
    
    // 4. 执行定时任务
    group.next().scheduleAtFixedRate(() -> { 
        System.out.println("ok");
    }, 0, 1, TimeUnit.SECONDS);

**处理 IO 事件：**

服务器端设置两个 EventLoop，即调用代码 `.group(new NioEventLoopGroup(2))`。

**运行结果如下图所示：**

可以看到两个 EventLoop 轮流处理 Channel，但 EventLoop 与 Channel 之间进行了绑定，一旦建立连接，这个 EventLoop 就会一直处理这个 Channel。

head、h1、tail 均表示 handler，默认情况下会额外的添加两个handler，即head 和 tail

![image-20210628100929235][]

**分工进一步细化：**

// 创建一个独立的EventLoopGroup，用来处理耗时较长的任务，防止handler执行时间过长而影响其他Channel的处理
    EventLoopGroup group = new DefaultEventLoopGroup();
    new ServerBootstrap()
    
        /** * 将EventLoop分为boss和worker * 参数1表示boss，只负责ServerSocketChannel上的accept事件，由于只有1个服务器channel，所以只占用一个线程 * 参数2表示worker，只负责socketChannel上的读写 */
        .group(new NioEventLoopGroup(), new NioEventLoopGroup(2))
        .channel(NioServerSocketChannel.class)
    
        .childHandler(new ChannelInitializer<NioSocketChannel>() { 
            @Override
            protected void initChannel(NioSocketChannel ch) throws Exception { 
    
                //参数1表示给这个handler起名字
                ch.pipeline().addLast("handler1", new ChannelInboundHandlerAdapter() { 
                    @Override
                    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
                        ByteBuf buf = (ByteBuf) msg;
                        ctx.fireChannelRead(msg); // 将结果传递给下一个handler
                    }
                })
    
                    //假设这个handler的执行时间较长，为了不影响其他channel的处理，不能使用当前线程
                    //使用三个参数，参数1表示handler执行时使用的是另一个的EventLoopGroup中的线程
                    //参数2表示给这个handler起名字
                    .addLast(group, "handler2", new ChannelInboundHandlerAdapter() { 
                        @Override
                        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
                            ByteBuf buf = (ByteBuf) msg;
                            //执行时间较长的任务处理
                        }
                    });
    
                //两个handler虽然按顺序处理，但使用的线程不同
            }
        })
        .bind(8080);

**执行情况如下图所示：**

![image-20210628101136093][]

**handler处理过程中如何切换线程？**

也就是为什么handler执行过程中可以将处理结果交给另一个线程，默认情况下是同一个线程走完handler的流程。

**关键代码：**

//一个handler调用下一个handler时会调用invokeChannelRead方法，沿着调用链调用handler
    static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) { 
        
        // 得到下一个handler的eventloop
        EventExecutor executor = next.executor();
        // EventLoop继承自EventExecutor
        
        // 判断下一个handler的线程是否与当前的handler的线程是同一个线程
        // 是，直接调用（同一个线程）
        if (executor.inEventLoop()) { 
            next.invokeChannelRead(m);
        } 
        
        // 不是，把要调用的代码封装为一个任务对象，由下一个handler的线程来调用
        // executor表示下一个handler的线程
        else { 
            executor.execute(new Runnable() { 
                @Override
                public void run() { 
                    next.invokeChannelRead(m);
                }
            });
        }
    }

### 2. Channel ###

channel 的主要作用：

*  `pipeline()` 方法添加处理器
 *  `write()` 方法将数据写入，但不是立刻写入（缓冲区满或者调用 `flush` 方法清空缓冲区才可真正写入）
 *  `writeAndFlush()` 方法将数据立刻写入

#### 2.1 ChannelFuture ####

这是一个客户端代码：

new Bootstrap()
        .group(new NioEventLoopGroup())
        .channel(NioSocketChannel.class)
        .handler(new ChannelInitializer<Channel>() { 
            @Override
            protected void initChannel(Channel ch) { 
                ch.pipeline().addLast(new StringEncoder());
            }
        })
        .connect("127.0.0.1", 8080) //此方法会返回一个ChannelFuture对象
        .sync()
        .channel()
        .writeAndFlush(new Date() + ": hello world!");

现在把它拆开来看：

*  返回的 ChannelFuture 对象的作用是利用 `channel()` 方法来获取 Channel 对象

注意：connect 方法是异步非阻塞的，意味着调用connect方法后，交给另一个线程去执行，此方法不等连接建立，方法执行就返回了，继续向下运行，如果此时连接还未建立，ChannelFuture 对象获得的 channel 是没有真正建立好连接的channel，导致数据无法发送。channel 是连接建立后才可用的。

故需要调用 `sync()` 方法阻塞直到连接建立。

**除了用 sync 方法可以让异步操作同步以外，还可以使用回调的方式：**

使用回调方法 `addListener` 异步进行连接结果的处理，使用 sync 方法，依然是主线程等待连接建立、进行结果的处理。而异步方式，等待连接建立、进行结果的处理都交由另一个线程，不需要主线程。

ChannelFuture channelFuture = new Bootstrap()
        .group(new NioEventLoopGroup())
        .channel(NioSocketChannel.class)
        .handler(new ChannelInitializer<Channel>() { 
            @Override
            protected void initChannel(Channel ch) { 
                ch.pipeline().addLast(new StringEncoder());
            }
        })
        .connect("127.0.0.1", 8080);
    System.out.println(channelFuture.channel()); // 1
    channelFuture.addListener(new ChannelFutureListener() { 
        @Override
        // 在连接建立好之后，会被建立连接的nio线程调用operationComplete方法
        public void operationComplete(ChannelFuture future) throws Exception { 
            System.out.println(future.channel()); // 2
        }
    });

*  执行到 1 时，连接未建立，打印 `[id: 0x749124ba]`
 *  ChannelFutureListener 会在连接建立时被建立连接的 NIO 线程调用，不是主线程调用，因此执行到 2 时，连接肯定建立了，打印 `[id: 0x749124ba, L:/127.0.0.1:57351 - R:/127.0.0.1:8080]`

#### 2.2 CloseFuture ####

*  `channel.close()` 用来关闭 channel，是一个异步非阻塞方法
 *  `channel.closeFuture()` 用来处理 channel 的关闭
    
     *  也就是关闭事件完成之后，做一些关闭后的处理
     *  `sync` 方法的作用是阻塞直到 channel 关闭
     *  `addListener` 方法是异步等待 channel 关闭

// 获取CloseFuture
    ChannelFuture closeFuture = channel.closeFuture();
    // CloseFuture也是属于ChannelFuture类
    
    //方式一，sync方法的作用是阻塞直到 channel 关闭
    closeFuture.sync();
    //处理关闭之后的操作;
    
    //方式二，addListener方法是异步等待channel关闭，由关闭线程调用operationComplete方法
    closeFuture.addListener(new ChannelFutureListener() { 
        @Override
        public void operationComplete(ChannelFuture future) throws Exception { 
    
            //关闭之后的操作
    
            //让NioEventLoopGroup中的所有线程停止，一般在连接关闭之后调用
            //此方法表示优雅的关闭，线程不接收新的任务，已经存在的任务处理完毕之后关闭
            group.shutdownGracefully();
        }
    });

#### 2.3 Netty为什么使用异步 ####

思考下面的场景，4 个医生给人看病，每个病人花费 20 分钟，而且是一个病人看完了，才能看下一个病人。假设病人源源不断的来，可以计算一下 4 个医生一天工作 8 小时，处理的病人总数是：`4 * 8 * 3 = 96`：

![image-20210628140243441][]

经研究发现，看病可以细分为四个步骤，经拆分后每个步骤需要 5 分钟，如下：

![image-20210628140335030][]

可以做如下优化，每个医生只负责其中一项任务。一开始，医生 2、3、4 分别要等待 5、10、15 分钟才能执行工作，但只要后续病人源源不断地来，他们就能够满负荷工作，每个线程（医生）5分钟就可以处理完一个病人（任务），效率几乎提升了4倍：

![image-20210628140423036][]

**注意：**

1.  单线程无法异步提升效率，必须配合多线程才能发挥异步的优势
2.  异步并没有缩短响应时间，反而因为一个任务需要进行不同线程的切换而导致响应时间有所增加
3.  提升的是吞吐量，即单位时间内处理任务的个数

### 3. Future & Promise ###

首先要说明 Netty 中的 Future 与 JDK 中的 Future 同名，但是是两个接口，Netty 的 Future 继承自 jdk 的 Future，而 Promise 又继承自 Netty 的 Future

*  JDK-Future 只能同步等待任务结束才能得到结果
 *  Netty-Future 可以同步等待任务结束得到结果，也可以异步方式得到结果

**上述二者都要等任务结束**

*  Netty-Promise 不仅有 Netty-Future 的功能，而且可以作为两个线程间传递结果的容器

<table> 
 <thead> 
  <tr> 
   <th>功能/名称</th> 
   <th>JDK-Future</th> 
   <th>Netty-Future</th> 
   <th>Netty-Promise</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>cancel</td> 
   <td>取消任务</td> 
   <td>-</td> 
   <td>-</td> 
  </tr> 
  <tr> 
   <td>isCanceled</td> 
   <td>任务是否取消</td> 
   <td>-</td> 
   <td>-</td> 
  </tr> 
  <tr> 
   <td>isDone</td> 
   <td>任务是否完成，不能区分成功失败</td> 
   <td>-</td> 
   <td>-</td> 
  </tr> 
  <tr> 
   <td>get</td> 
   <td>获取任务结果，阻塞等待</td> 
   <td>-</td> 
   <td>-</td> 
  </tr> 
  <tr> 
   <td>getNow</td> 
   <td>-</td> 
   <td>获取任务结果，非阻塞，还未产生结果时返回 null</td> 
   <td>-</td> 
  </tr> 
  <tr> 
   <td>await</td> 
   <td>-</td> 
   <td>等待任务结束，如果任务失败，不会抛异常，判断成功还是失败需要自己调用 isSuccess 判断</td> 
   <td>-</td> 
  </tr> 
  <tr> 
   <td>sync</td> 
   <td>-</td> 
   <td>等待任务结束，如果任务失败，抛出异常，此方法仅等待任务结束，不获取结果，而get方法要获取结果</td> 
   <td>-</td> 
  </tr> 
  <tr> 
   <td>isSuccess</td> 
   <td>-</td> 
   <td>判断任务是否成功</td> 
   <td>-</td> 
  </tr> 
  <tr> 
   <td>cause</td> 
   <td>-</td> 
   <td>获取失败信息，非阻塞，如果没有失败，返回null</td> 
   <td>-</td> 
  </tr> 
  <tr> 
   <td>addLinstener</td> 
   <td>-</td> 
   <td>添加回调，异步接收结果</td> 
   <td>-</td> 
  </tr> 
  <tr> 
   <td>setSuccess</td> 
   <td>-</td> 
   <td>-</td> 
   <td>设置成功结果</td> 
  </tr> 
  <tr> 
   <td>setFailure</td> 
   <td>-</td> 
   <td>-</td> 
   <td>设置失败结果</td> 
  </tr> 
 </tbody> 
</table>

#### 3.1 JDK-Future ####

// 1. 线程池
    ExecutorService service = Executors.newFixedThreadPool(2);
    
    // 2. 提交任务
    Future<Integer> future = service.submit(new Callable<Integer>() { 
        @Override
        public Integer call() throws Exception { 
            Thread.sleep(1000);
            return 50;
        }
    });
    
    // 3. 主线程通过future来获取结果
    System.out.println(future.get()); //阻塞等待获取结果
    
    //线程池中的线程执行任务，主线程阻塞等待任务结束

#### 3.2 Netty-Future ####

NioEventLoopGroup group = new NioEventLoopGroup();
    EventLoop eventLoop = group.next(); //取出一个线程
    
    //返回的是netty包下的future对象
    Future<Integer> future = eventLoop.submit(new Callable<Integer>() { 
        @Override
        public Integer call() throws Exception { 
            Thread.sleep(1000);
            return 70;
        }
    });
    
    // future.get(); 同步阻塞等待获取结果
    
    // 异步方式
    future.addListener(new GenericFutureListener<Future<? super Integer>>(){ 
    
        //eventloop线程计算出结果后，调用此方法
        @Override
        public void operationComplete(Future<? super Integer> future) throws Exception { 
            System.out.println(future.getNow());
            //调用get方法亦可，回调方法被调用，结果一定计算出来了
        }
    });

#### 3.3 Netty-Promise ####

// 1. 准备EventLoop线程对象
    EventLoop eventLoop = new NioEventLoopGroup().next();
    
    // 2. 主动创建promise, 泛型表示容器中数据的类型
    // 参数指定EventLoop对象
    DefaultPromise<Integer> promise = new DefaultPromise<>(eventLoop);
    
    // 3. 启动任意一个线程执行计算，计算完毕后向promise容器添加结果
    new Thread(() -> { 
        try { 
            int i = 1 / 0; //手动制造异常
            promise.setSuccess(80); //执行正确时装入的结果
        } catch (Exception e) { 
            e.printStackTrace();
            promise.setFailure(e); //出错时，指定异常原因，参数是Throwable cause
        }
    
    }).start();
    
    // 4. 主线程接收结果
    System.out.println(promise.get());
    //阻塞直到获取结果，亦可使用异步方式
    }

### 4. Handler & Pipeline ###

Handler 用来处理 Channel 上的各种事件，分为入站、出站两种。所有 Handler 被连成一串，就是 Pipeline

*  入站处理器通常是 ChannelInboundHandlerAdapter 的子类，主要用来读取客户端数据，写回结果
 *  出站处理器通常是 ChannelOutboundHandlerAdapter 的子类，主要对写回结果进行加工

打个比喻，每个 Channel 是一个产品的加工车间，Pipeline 是车间中的流水线，Handler 就是流水线上的各道工序，而后面要讲的 ByteBuf 是原材料，需要经过很多工序的加工：先经过一道道入站工序，再经过一道道出站工序最终变成产品。

**Handler的执行顺序：**

服务器端代码：

new ServerBootstrap()
        .group(new NioEventLoopGroup())
        .channel(NioServerSocketChannel.class)
        .childHandler(new ChannelInitializer<NioSocketChannel>() { 
            @Override
            protected void initChannel(NioSocketChannel ch) throws Exception { 
    
                // 1. 通过channel拿到pipeline
                ChannelPipeline pipeline = ch.pipeline();
    
                // 2. 添加处理器，addLast方法将新添加的处理器添加到最后
                // 默认情况下，会额外的添加两个handler，head和tail，也就是新添加的handler都在tail之前
                // 添加3个入站处理器，监听read事件，参数1表示给这个handler起名字
                pipeline.addLast("h1", new ChannelInboundHandlerAdapter(){ 
                    @Override
                    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
                        log.debug("1");
                        super.channelRead(ctx, msg);
                        // 将数据传递给下一个入站handler，如果不调用，调用链会断
                        // 参数2表示经过此handler处理后的数据传递给下一个handler的msg参数
                    }
                });
    
                pipeline.addLast("h2", new ChannelInboundHandlerAdapter(){ 
                    @Override
                    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
                        log.debug("2");
                        super.channelRead(ctx, msg);
                    }
                });
    
                pipeline.addLast("h3", new ChannelInboundHandlerAdapter(){ 
                    @Override
                    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
                        log.debug("3");
    
                        //向channel中写入数据，write事件才会被触发
                        ch.writeAndFlush(ctx.alloc().buffer().writeBytes("server...".getBytes()));
                        //ctx.writeAndFlush(ctx.alloc().buffer().writeBytes("server...".getBytes()));
    
                        //两种方式写入数据后的处理器调用情况不同，后续详细讲述
    
                        //注意，此处无需调用super.channelRead(ctx, name);
                        //因为没有下一个入站处理器，无需调用
                    }
                });
    
                //执行到此，处理器的顺序为head、h1、h2、h3、tail，是一个双向链表
    
                //添加三个出站处理器，监听write事件
                pipeline.addLast("h4", new ChannelOutboundHandlerAdapter(){ 
                    @Override
                    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { 
                        log.debug("4");
                        super.write(ctx, msg, promise);
                        //调用下一个应该被调用的出站处理器
                    }
                });
    
                pipeline.addLast("h5", new ChannelOutboundHandlerAdapter(){ 
                    @Override
                    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { 
                        log.debug("5");
                        super.write(ctx, msg, promise);
                    }
                });
    
                pipeline.addLast("h6", new ChannelOutboundHandlerAdapter(){ 
                    @Override
                    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { 
                        log.debug("6");
                        super.write(ctx, msg, promise);
                    }
                });
    
                //执行到此，处理器的顺序为 head、h1、h2、h3、h4、h5、h6、tail，是一个双向链表
            }
        })
        .bind(8080);

**运行结果：**

1
    2
    3
    6
    5
    4
    //入站处理器是顺序的，出站处理器是逆序的

可以看到，入站处理器是按照 addLast 的顺序执行的，而出站处理器是按照 addLast 的逆序执行的。ChannelPipeline 的实现是一个 ChannelHandlerContext（包装了 ChannelHandler） 组成的双向链表：

![image-20210628140622794][]

*  入站处理器中，`ctx.fireChannelRead(msg)` 是 **调用下一个入站处理器**
 *  最后一个入站处理器调用 `channel().write(msg)` 会 **从尾部开始逆序找出站处理器**，也就是上述运行结果：6、5、4
 *  最后一个入站处理器调用 `ctx.write(msg, promise)` 会 **从当前处理器逆序找出站处理器**
 *  `channel().write(msg)` 对比 `ctx.write(msg)`
    
     *  都是触发出站处理器的执行
     *  channel().write(msg) 从尾部开始查找出站处理器
     *  ctx.write(msg) 是从当前节点找上一个出站处理器
     *  最后一个入站处理器如果调用 `ctx.write(msg)` 仅会打印 1 2 3，因为此方法是是从当前节点找上一个出站处理器，节点 3 之前没有其它出站处理器了
     *  最后一个入站处理器如果调用 `channel().write(msg)` 会打印 1 2 3 6 5 4，因为此方法是从尾部开始查找出站处理器

### 5. ByteBuf ###

是对字节数据的封装

**1）创建**

ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(10); 
    //如果不指定参数，则默认容量是256，可动态扩容

上面代码创建了一个默认的 ByteBuf（池化基于直接内存的 ByteBuf），初始容量是 10 个字节

**输出：**

PooledUnsafeDirectByteBuf(ridx: 0, widx: 0, cap: 10)

从输出结果可以看出：

*  Pooled表示使用池化
 *  Direct表示使用直接内存
 *  ridx读指针索引为0，widx写指针索引为0
 *  cap表示容量

下面讲解这几个概念：

**2）直接内存 vs 堆内存**

堆内存分配效率高，读写效率低，并且由于涉及到回收机制，数据的搬迁操作也会影响效率。

直接内存分配效率低，读写效率高（Java内存和系统内存共用一块空间，减少了一次复制操作），并且不受 JVM 垃圾回收的管理，但也要注意及时主动释放。

可以使用下面的代码来创建池化基于堆的 ByteBuf：

ByteBuf buffer = ByteBufAllocator.DEFAULT.heapBuffer(10);

可以使用下面的代码来创建池化基于直接内存的 ByteBuf（不指定默认使用的也是直接内存）：

ByteBuf buffer = ByteBufAllocator.DEFAULT.directBuffer(10);

**3）池化 vs 非池化**

池化的最大意义在于可以重用 ByteBuf，优点有：

*  没有池化，则每次都得创建新的 ByteBuf 实例，这个操作对于直接内存代价昂贵，就算是堆内存，也会增加 GC 压力
 *  有了池化，则可以重用池中 ByteBuf 实例，并且采用了高效的内存分配算法提升分配效率
 *  高并发时，池化功能更节约内存，减少内存溢出的可能

池化功能是否开启，可以通过下面的系统环境变量来设置：

-Dio.netty.allocator.type={ unpooled|pooled}

**4）组成**

ByteBuf 由四部分组成：

*  废弃字节
 *  可读字节
 *  可写字节
 *  可扩容字节

ByteBuf 有四个属性：

*  读指针
 *  写指针
 *  容量
 *  最大容量（默认的最大容量是整数的最大值）

![image-20210626174727092][]

最开始读写指针都在 0 位置，写入数据写指针会向后移动，读取数据读指针会向后移动，读取后的部分称为废弃字节。

**对比 NIO 的 ByteBuffer：**

*  NIO 的 ByteBuffer，读写指针使用的是同一个，读写过程必须得切换读写模式，不如 Netty 的 ByteBuf 方便
 *  Netty 的 ByteBuf 支持动态扩容，而 NIO 的 ByteBuffer 不支持

**5）写入**

写入 ByteBuf 时常用的方法：

<table> 
 <thead> 
  <tr> 
   <th>方法签名</th> 
   <th>含义</th> 
   <th>备注</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>writeBoolean(boolean value)</td> 
   <td>写入 boolean 值</td> 
   <td>用一个字节 01|00 代表 true|false</td> 
  </tr> 
  <tr> 
   <td>writeByte(int value)</td> 
   <td>写入 byte 值</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>writeShort(int value)</td> 
   <td>写入 short 值</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>writeInt(int value)</td> 
   <td>写入 int 值</td> 
   <td>Big Endian（大端写入），即 0x250，写入后 00 00 02 50，先写高位，再写低位</td> 
  </tr> 
  <tr> 
   <td>writeIntLE(int value)</td> 
   <td>写入 int 值</td> 
   <td>Little Endian（小端写入），即 0x250，写入后 50 02 00 00，先写低位，再写高位</td> 
  </tr> 
  <tr> 
   <td>writeLong(long value)</td> 
   <td>写入 long 值</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>writeChar(int value)</td> 
   <td>写入 char 值</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>writeFloat(float value)</td> 
   <td>写入 float 值</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>writeDouble(double value)</td> 
   <td>写入 double 值</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>writeBytes(ByteBuf src)</td> 
   <td>写入 netty 的 ByteBuf</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>writeBytes(byte[] src)</td> 
   <td>写入 byte[]</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>writeBytes(ByteBuffer src)</td> 
   <td>写入 nio 的 ByteBuffer</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>int writeCharSequence(CharSequence sequence, Charset charset)</td> 
   <td>写入字符串</td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

**注意：**

*  上述的所有方法都会使写指针后移

*  这些方法中未指明返回值的，其返回值都是 ByteBuf，意味着可以链式调用
 *  网络传输中默认习惯写入的方式为大端写入
 *  还有一类 set 开头的一系列方法，也可以写入数据，但不会改变写指针的位置

**6）扩容**

扩容规则如下：

*  如果写入数据后大小未超过 512 字节，则选择下一个 16 的整数倍为容量，例如写入后大小为 12 字节 ，则扩容后 capacity 是 16 字节
 *  如果写入数据后大小超过 512 字节，则选择下一个 2^n 为容量，例如写入后大小为 513 字节，则扩容后 capacity 是 2^10=1024 字节（2^9=512 已经不够了）
 *  扩容不能超过最大容量，即整数的最大值，如果超过会报错

**7）读取**

读取一个字节：`buf.readByte()`，读指针后移。

读取一个整数（4个字节）：`buf.readInt()`，读指针后移。

读过的内容，就属于废弃部分了，再读只能读那些尚未读取的部分。

**如果需要重复读取：**

//可以在read前先做个标记mark
    buffer.markReaderIndex();
    
    //读指针重置到标记位置
    buffer.resetReaderIndex();

还有一种办法是采用 get 开头的一系列方法，这些方法不会改变读指针的位置。

**8）retain & release**

*  UnpooledHeapByteBuf 使用的是 JVM 内存，只需等 GC 回收内存即可
 *  UnpooledDirectByteBuf 使用的是直接内存，需要手动调用操作系统底层的方法来回收内存
 *  PooledByteBuf 和它的子类使用了池化机制，需要更复杂的规则来回收内存，并不是真正的回收内存，而是将内存还回内存池

Netty 采用引用计数法来控制回收内存，每个 ByteBuf 都实现了 `ReferenceCounted` 接口，此接口提供通用的方法实现不同 ByteBuf 的回收

*  每个 ByteBuf 对象的初始计数为 1，表示有人在使用，不能回收
 *  调用 `release` 方法计数减 1，如果计数为 0，ByteBuf 内存被回收
    
     *  当计数为 0 时，根据不同的 ByteBuf 的实现，进行不同的回收处理
 *  调用 `retain` 方法计数加 1

因为 pipeline 的存在，一般需要将 ByteBuf 传递给下一个 Handler，如果在某个handler的 finally 中 release 了，就失去了传递性（当然，如果在某个 Handler 内这个 ByteBuf 已完成了它的使命，那么便无须再传递）

基本规则是，**哪个Handler是 ByteBuf 的最后使用者，则谁负责 release**，详细分析如下：

*  入站 ByteBuf 的处理原则
    
     *  对原始 ByteBuf 不做处理，调用 `ctx.fireChannelRead(msg)` 向后传递，这时无须 release
     *  将原始 ByteBuf 转换为其它类型的 Java 对象，这时 ByteBuf 就没用了，必须 release
     *  如果不调用 `ctx.fireChannelRead(msg)` 向后传递，那么也必须 release
     *  注意各种异常，如果 ByteBuf 没有成功传递到下一个 Handler，必须 release
     *  假设消息一直向后传，那么默认提供的 tail 处理器会负责释放未处理的消息
 *  出站 ByteBuf 的处理原则
    
     *  出站消息最终都会转换为 ByteBuf 输出，如果在出站 handler 中没有释放，会一直向前传到 head 处理器，由默认提供的 head 处理器 release
 *  如果中间的 handler 都没有 release，则传递到头尾后由默认提供的 head、tail 处理器 release
    
     *  前提是未释放的 ByteBuf 能传递到头尾，如果传递到头尾的是 ByteBuf 转换后的数据类型，则不会释放

#### 5.1 零拷贝 ####

##### 5.1.1 slice #####

对原始 ByteBuf 切片成多个 ByteBuf，切片后的 ByteBuf 并没有发生内存复制（仅仅是在逻辑上划分成多个ByteBuf），还是使用原始 ByteBuf 的物理内存，切片后的 ByteBuf 维护独立的读写指针：

![image-20210628141024838][]

**代码演示：**

ByteBuf buf = ByteBufAllocator.DEFAULT.buffer(10);
    buf.writeBytes(new byte[]{ 'a','b','c','d','e','f','g','h','i','j'});
    
    // 在切片过程中，没有发生数据复制，切片仍然用的是原内容的物理内存
    // 带参slice的参数1表示从原始数据的哪个索引开始切片，第二个参数表示切片的内容是多长
    // 无参slice是从原始ByteBuf的读指针到写指针之间的内容进行切片
    
    ByteBuf f1 = buf.slice(0, 5);
    
    // 切片之后的最大容量被固定为这个区间的大小，不能向切片中添加数据，如果添加，会报错IndexOutOfBoundsException
    
    /** * 当原始数据写入新数据时，切片的内容也会被修改，因为使用的是同一块物理内存 * 当原始数据读出数据时，切片的读指针不会移动，因为切片拥有独立的读写指针 */
    
    f1.retain();
    // 切片的引用计数+1，如果不这样做的话，原始内容调用release，切片内容会消失
    
    ByteBuf f2 = buf.slice(5, 5);
    f2.retain();
    
    // 切片的数据最好由自己来释放
    f1.release();
    f2.release();

##### 5.1.2 duplicate #####

截取原始 ByteBuf 的所有内容，并且没有最大容量的限制，也是与原始 ByteBuf 使用同一块物理内存，只是读写指针是独立的：

![image-20210628141120525][]

##### 5.1.3 CompositeByteBuf #####

将多个 ByteBuf 合并为一个逻辑上的 ByteBuf，且不会发生数据的复制

**代码演示：**

// 创建两个将要合并的ByteBuf
    ByteBuf buf1 = ByteBufAllocator.DEFAULT.buffer();
    buf1.writeBytes(new byte[]{ 1, 2, 3, 4, 5});
    
    ByteBuf buf2 = ByteBufAllocator.DEFAULT.buffer();
    buf2.writeBytes(new byte[]{ 6, 7, 8, 9, 10});
    
    // 创建合并后的ByteBuf，使用CompositeByteBuf对象
    CompositeByteBuf buffer = ByteBufAllocator.DEFAULT.compositeBuffer();
    
    // 组合两个ByteBuf到新的ByteBuf中
    // 参数1表示写入数据时写指针自动增长，如果写指针的位置不自动调整，则数据无法写入
    buffer.addComponents(true, buf1, buf2);

CompositeByteBuf 是一个组合的 ByteBuf，它内部维护了一个 Component 数组，每个 Component 管理一个 ByteBuf，记录了这个 ByteBuf 相对于整体偏移量等信息，代表着整体中某一段的数据

##### 5.1.4 Unpooled #####

Unpooled 是一个工具类，提供了非池化的 ByteBuf 创建、组合、复制等操作。

这里仅介绍其与零拷贝相关的 `wrappedBuffer()` 方法，可以用来包装 ByteBuf。

**代码演示：**

// 创建两个待包装的ByteBuf
    ByteBuf buf1 = ByteBufAllocator.DEFAULT.buffer(5);
    buf1.writeBytes(new byte[]{ 1, 2, 3, 4, 5});
    
    ByteBuf buf2 = ByteBufAllocator.DEFAULT.buffer(5);
    buf2.writeBytes(new byte[]{ 6, 7, 8, 9, 10});
    
    // 将多个ByteBuf包装成一个新的ByteBuf
    // 当包装ByteBuf个数超过一个时, 底层使用了CompositeByteBuf，避免了数据的复制
    ByteBuf buf3 = Unpooled.wrappedBuffer(buf1, buf2);

##### 5.1.5 copy #####

与零拷贝对立，调用 `copy()` 方法会将底层内存数据进行深拷贝，因此无论读写，都与原始 ByteBuf 无关

#### 5.2 ByteBuf优势 ####

*  池化可以重用池中的 ByteBuf 实例，更节约内存，减少内存溢出的可能
 *  读写指针分离，不需要像 NIO 的 ByteBuffer 一样切换读写模式
 *  可以自动扩容
 *  代码支持链式调用，使用更流畅
 *  很多地方体现零拷贝，例如 slice、duplicate、CompositeByteBuf

## 三、消息边界问题 ##

### 1. 现象分析 ###

**粘包**

*  现象，发送 abc def，接收 abcdef
 *  原因
    
     *  应用层：接收方 ByteBuf 设置太大（Netty 默认 1024 字节）
        
         *  `serverBootstrap.option(ChannelOption.SO_RCVBUF, 10)` 影响底层接收缓冲区的大小
     *  滑动窗口：假设发送方 256 bytes 表示一个完整报文，但由于接收方处理不及时且窗口大小过大，这 256 bytes 字节就会缓冲在接收方的滑动窗口中，当滑动窗口中缓冲了多个报文就会粘包

**半包**

*  现象，发送 abcdef，接收 abc def
 *  原因
    
     *  应用层：接收方 ByteBuf 小于实际发送的数据量
     *  滑动窗口：假设接收方的窗口只剩了 128 bytes，发送方的报文大小是 256 bytes，这时放不下了，只能先发送前 128 bytes，等待 ack 后才能发送剩余部分，这就造成了半包

出现上述两种情况的本质是因为 TCP 是流式协议，消息无边界，为了避免上述情况，必须自己去找出消息的边界

### 2. 解决方案 ###

1.  短链接，每发一个消息建立一次连接，这样连接建立到连接断开之间就是消息的边界，缺点是效率太低
2.  客户端和服务器约定消息的长度，每一条消息采用固定长度，缺点是浪费空间
3.  每一条消息采用分隔符，例如 `\n`，缺点是需要遍历找到分隔符
4.  使用定长字节表示内容长度 + 实际内容

**方法1：短链接**

半包情况用这种办法并不好，因为接收方的缓冲区大小是有限的，可能发送的数据容量比接收缓冲区大（接收缓冲区大小默认是1024个字节）

**方法2：固定长度**

当遇到半包问题时，等待消息的继续发送，直到消息的长度达到约定的长度。

当遇到黏包问题时，把满足约定长度的那部分进行处理，剩余的内容不处理，与后来的信息做拼接，直到满足约定的长度。

让所有数据包长度固定（假设长度为 8 字节），服务器端加入如下代码：

ch.pipeline().addLast(new FixedLengthFrameDecoder(8));

缺点是数据包的大小不好把握

*  长度定的太大，浪费空间
 *  长度定的太小，对某些数据包又显得不够

**方法3：使用分隔符**

服务端加入如下代码，默认以 `\n` 或 `\r\n` 作为分隔符，如果超出指定长度仍未出现分隔符，则抛出异常：

//以换行符 \n 或 \r\n 作为分隔符
    //参数指定最大长度，如果接收的消息达到了最大长度但是还没有找到换行符，则抛出异常
    ch.pipeline().addLast(new LineBasedFrameDecoder(1024));

缺点：

1.  处理字符数据比较合适，但如果内容本身包含了分隔符，那么就会解析错误
2.  效率较低，需要遍历消息寻找分隔符

**方法4：设定协议**

在发送消息前，先约定客户端发送的消息所要满足的格式和长度：

//参数1表示最大长度，当接收的消息超过了最大长度还没有发现分割标准，则认为失败
    ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(1024, 0, 1, 0, 1));

**其余四个参数的介绍：**

![image-20210628185914129][]

**情况一：**

![image-20210628185937903][image-20210628185914129]

发送的消息长度是12个字节，再加上表示消息长度的Length字段，消息的总长度变为14个字节。参数1表示从哪个位置开始表示Length字段，参数2表示Length字段的长度，则剩余部分都是消息的长度

**情况二：**

不想要Length字段，最终的结果要将Length字段剥离出去：

![image-20210628190245033][image-20210628185914129]

参数4表示从头剥离几个字节

**情况三：**

![image-20210628190745499][image-20210628185914129]

在Length字段和消息内容之间添加了附加内容Header，但Length字段仅表示消息内容的长度，不包含附加内容的长度，参数3表示从Length字段到消息内容之间隔了几个字节（假设Header字段的长度为2）

**情况四：**

![image-20210628191122441][image-20210628185914129]

[NIO_ByteBuffer]: https://blog.csdn.net/weixin_49343190/article/details/118099988
[NIO_Selector]: https://blog.csdn.net/weixin_49343190/article/details/118294271
[image-20210628135710527]: /images/20221005/c680f34e8c7646bab8d219f03364a09e.png
[image-20210625162800228]: /images/20221005/e40a2f23ee954880b1f07f1a6c78aec2.png
[image-20210628100929235]: /images/20221005/f44978ad36a2421ca8c9d74ad3f83c82.png
[image-20210628101136093]: /images/20221005/c4431bda188444a1ba6a25099dfcdde4.png
[image-20210628140243441]: /images/20221005/463a93a181834b7583a174a9a1d77d89.png
[image-20210628140335030]: /images/20221005/41dee977df1945bc85b8170b1c8c19d8.png
[image-20210628140423036]: /images/20221005/9133f1140cfb4988ad24d1251f6478a6.png
[image-20210628140622794]: /images/20221005/f9e35b2cfb734226b14d4696bea0c819.png
[image-20210626174727092]: /images/20221005/c6edffe8362b454eb88cc206b60582c5.png
[image-20210628141024838]: /images/20221005/e1c452d321264d67987445ae525e75dd.png
[image-20210628141120525]: /images/20221005/d425dc788cce43da80792bd9dc5871d3.png
[image-20210628185914129]: /images/20221005/50ff24ee6a8b4a2b93571e764d0d1953.png