Netty原理与基础（三）

1.Pipeline流水线

Netty的业务处理器流水线ChannelPipeline是基于责任链设计模式（Chain of Responsibility）来设计的，内部是一个双向链表结构，能够支持动态地添加和删除Handler业务处理器。

入站操作

image.png
出站操作
出站流水从后往前

image.png

1.1ChanneHandlerContext上下文

不论定义那种类型的Handler业务处理器，最终形式都是以双向链表方式保存的
在Handler业务处理器被添加到流水线中时，会创建一个通道处理器上下文ChannelHandlerContext，它代表了ChannelHandler通道处理器和ChannelPipeline通道流水线之间的关联。
主要分类
1.获取上下文所关联的Netty组件实例，如所关联的通道、所关联的流水线、上下文内部Handler业务处理器实例等；

是入站和出站处理方法

Channel、Handler、ChannelHandlerContext三者的关系为：
Channel通道拥有一条ChannelPipeline通道流水线，每一个流水线节点为一个ChannelHandlerContext通道处理器上下文对象，每一个上下文中包裹了一个ChannelHandler通道处理器。在ChannelHandler通道处理器的入站/出站处理方法中，Netty都会传递一个Context上下文实例作为实际参数。通过Context实例的实参，在业务处理中，可以获取ChannelPipeline通道流水线的实例或者Channel通道的实例。

1.2阶段流水线的处理方式

1.入站在channelRead方法中，不去调用父类的channelRead入站方法
-2.出站处理流程中，只要开始就不能被阶段

2.ByteBuf缓冲区

ByteBuf的优势：
- Pooling (池化，这点减少了内存复制和GC，提升了效率)
- 复合缓冲区类型，支持零复制
- 不需要调用flip()方法去切换读/写模式
- 扩展性好，例如StringBuffer
- 可以自定义缓冲区类型
- 读取和写入索引分开
- 方法的链式调用
- 可以进行引用计数，方便重复使用
ByteBuf是一个字节容器，内部是一个字节数组

image.png

2.1ByteBuf的重要属性

image.png

readerIndex（读指针）：指示读取的起始位置。每读取一个字节，readerIndex自动增加1。一旦readerIndex与writerIndex相等，则表示ByteBuf不可读了。
writerIndex（写指针）：指示写入的起始位置。每写一个字节，writerIndex自动增加1。一旦增加到writerIndex与capacity()容量相等，则表示ByteBuf已经不可写了。capacity()是一个成员方法，不是一个成员属性，它表示ByteBuf中可以写入的容量。注意，它不是最大容量maxCapacity。
maxCapacity（最大容量）：表示ByteBuf可以扩容的最大容量。当向ByteBuf写数据的时候，如果容量不足，可以进行扩容。扩容的最大限度由maxCapacity的值来设定，超过maxCapacity就会报错。

2.2ByteBuf的三组方法

容量系列：capacity()、maxCapacity()
写入系列: isWritable()、writableBytes() 等
读取系列： isReadable( )、 readableBytes( )等

2.3代码示例

public class WriteReadTest {

    @Test
    public void testWriteRead() {
        ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer();
        print("动作：分配 ByteBuf(9, 100)", buffer);
        buffer.writeBytes(new byte[]{1, 2, 3, 4});
        print("动作：写入4个字节 (1,2,3,4)", buffer);
        Logger.info("start==========:get==========");
        getByteBuf(buffer);
        print("动作：取数据 ByteBuf", buffer);
        Logger.info("start==========:read==========");
        readByteBuf(buffer);
        print("动作：读完 ByteBuf", buffer);
    }

    //读取一个字节
    private void readByteBuf(ByteBuf buffer) {
        while (buffer.isReadable()) {
            Logger.info("读取一个字节:" + buffer.readByte());
        }
    }


    //读取一个字节，不改变指针
    private void getByteBuf(ByteBuf buffer) {
        for (int i = 0; i < buffer.readableBytes(); i++) {
            Logger.info("读取一个字节:" + buffer.getByte(i));
        }
    }

}

日志打印：

[main|PrintAttribute.print] |>  after ===========动作：分配 ByteBuf(9, 100)============ 
[main|PrintAttribute.print] |>  1.0 isReadable(): false 
[main|PrintAttribute.print] |>  1.1 readerIndex(): 0 
[main|PrintAttribute.print] |>  1.2 readableBytes(): 0 
[main|PrintAttribute.print] |>  2.0 isWritable(): true 
[main|PrintAttribute.print] |>  2.1 writerIndex(): 0 
[main|PrintAttribute.print] |>  2.2 writableBytes(): 256 
[main|PrintAttribute.print] |>  3.0 capacity(): 256 
[main|PrintAttribute.print] |>  3.1 maxCapacity(): 2147483647 
[main|PrintAttribute.print] |>  3.2 maxWritableBytes(): 2147483647 
[main|PrintAttribute.print] |>  after ===========动作：写入4个字节 (1,2,3,4)============ 
[main|PrintAttribute.print] |>  1.0 isReadable(): true 
[main|PrintAttribute.print] |>  1.1 readerIndex(): 0 
[main|PrintAttribute.print] |>  1.2 readableBytes(): 4 
[main|PrintAttribute.print] |>  2.0 isWritable(): true 
[main|PrintAttribute.print] |>  2.1 writerIndex(): 4 
[main|PrintAttribute.print] |>  2.2 writableBytes(): 252 
[main|PrintAttribute.print] |>  3.0 capacity(): 256 
[main|PrintAttribute.print] |>  3.1 maxCapacity(): 2147483647 
[main|PrintAttribute.print] |>  3.2 maxWritableBytes(): 2147483643 
[main|WriteReadTest.testWriteRead] |>  start==========:get========== 
[main|WriteReadTest.getByteBuf] |>  读取一个字节:1 
[main|WriteReadTest.getByteBuf] |>  读取一个字节:2 
[main|WriteReadTest.getByteBuf] |>  读取一个字节:3 
[main|WriteReadTest.getByteBuf] |>  读取一个字节:4 
[main|PrintAttribute.print] |>  after ===========动作：取数据 ByteBuf============ 
[main|PrintAttribute.print] |>  1.0 isReadable(): true 
[main|PrintAttribute.print] |>  1.1 readerIndex(): 0 
[main|PrintAttribute.print] |>  1.2 readableBytes(): 4 
[main|PrintAttribute.print] |>  2.0 isWritable(): true 
[main|PrintAttribute.print] |>  2.1 writerIndex(): 4 
[main|PrintAttribute.print] |>  2.2 writableBytes(): 252 
[main|PrintAttribute.print] |>  3.0 capacity(): 256 
[main|PrintAttribute.print] |>  3.1 maxCapacity(): 2147483647 
[main|PrintAttribute.print] |>  3.2 maxWritableBytes(): 2147483643 
[main|WriteReadTest.testWriteRead] |>  start==========:read========== 
[main|WriteReadTest.readByteBuf] |>  读取一个字节:1 
[main|WriteReadTest.readByteBuf] |>  读取一个字节:2 
[main|WriteReadTest.readByteBuf] |>  读取一个字节:3 
[main|WriteReadTest.readByteBuf] |>  读取一个字节:4 
[main|PrintAttribute.print] |>  after ===========动作：读完 ByteBuf============ 
[main|PrintAttribute.print] |>  1.0 isReadable(): false 
[main|PrintAttribute.print] |>  1.1 readerIndex(): 4 
[main|PrintAttribute.print] |>  1.2 readableBytes(): 0 
[main|PrintAttribute.print] |>  2.0 isWritable(): true 
[main|PrintAttribute.print] |>  2.1 writerIndex(): 4 
[main|PrintAttribute.print] |>  2.2 writableBytes(): 252 
[main|PrintAttribute.print] |>  3.0 capacity(): 256 
[main|PrintAttribute.print] |>  3.1 maxCapacity(): 2147483647 
[main|PrintAttribute.print] |>  3.2 maxWritableBytes(): 2147483643

使用get取数据不会影想ByteBuf的指针

2.4ByteBuf的引用计数

Netty的ByteBuf的内存回收工作是通过引用计数的方式管理的
Netty采用“计数器”来追踪ByteBuf的生命周期，一是对PooledByteBuf的支持，二是能够尽快地“发现”那些可以回收的ByteBuf（非Pooled），以便提升ByteBuf的分配和销毁的效率
什么是Pooled（池化）的ByteBuf缓冲区呢？

在通信程序的执行过程中，Buffer缓冲区实例会被频繁创建、使用、释放。大家都知道，频繁创建对象、内存分配、释放内存，系统的开销大、性能低，如何提升性能、提高Buffer实例的使用率呢？从Netty4版本开始，新增了对象池化的机制。即创建一个Buffer对象池，将没有被引用的Buffer对象，放入对象缓存池中；当需要时，则重新从对象缓存池中取出，而不需要重新创建

引用计数的大致规则如下：

在默认情况下，当创建完一个ByteBuf时，它的引用为1；每次调用retain()方法，它的引用就加1；每次调用release()方法，就是将引用计数减1；如果引用为0，再次访问这个ByteBuf对象，将会抛出异常；如果引用为0，表示这个ByteBuf没有哪个进程引用它，它占用的内存需要回收。

在Netty中，引用计数为0的缓冲区不能再继续使用

image.png
为了确保引用计数不会混乱，在Netty的业务处理器开发过程中，应该坚持一个原则：retain和release方法应该结对使用。
当引用计数已经为0, Netty会进行ByteBuf的回收。分为两种情况：

（1）Pooled池化的ByteBuf内存，回收方法是：放入可以重新分配的ByteBuf池子，等待下一次分配。（2）Unpooled未池化的ByteBuf缓冲区，回收分为两种情况：如果是堆（Heap）结构缓冲，会被JVM的垃圾回收机制回收；如果是Direct类型，调用本地方法释放外部内存（unsafe.freeMemory）

2.5 ByteBuf的Allocator分配器

Netty通过ByteBufAllocator分配器来创建缓冲区和分配内存空间。Netty提供了ByteBufAllocator的两种实现：PoolByteBufAllocator和UnpooledByteBufAllocator。

PoolByteBufAllocator（池化ByteBuf分配器）将ByteBuf实例放入池中，提高了性能，将内存碎片减少到最小；这个池化分配器采用了jemalloc高效内存分配的策略，该策略被好几种现代操作系统所采用。
UnpooledByteBufAllocator是普通的未池化ByteBuf分配器，它没有把ByteBuf放入池中，每次被调用时，返回一个新的ByteBuf实例；通过Java的垃圾回收机制回收。
内存管理的策略可以灵活调整，这是使用Netty所带来的又一个好处。只需一行简单的配置，就能获得到池化缓冲区带来的好处

public class AllocatorTest {
    @Test
    public void showAlloc() {
        ByteBuf buffer = null;
        //方法一：默认分配器，分配初始容量为9，最大容量100的缓冲
        buffer = ByteBufAllocator.DEFAULT.buffer(9, 100);
        //方法二：默认分配器，分配初始为256，最大容量Integer.MAX_VALUE 的缓冲
        buffer = ByteBufAllocator.DEFAULT.buffer();
        //方法三：非池化分配器，分配基于Java的堆内存缓冲区
        buffer = UnpooledByteBufAllocator.DEFAULT.heapBuffer();
        //方法四：池化分配器，分配基于操作系统的管理的直接内存缓冲区
        buffer = PooledByteBufAllocator.DEFAULT.directBuffer();
        //…..其他方法

    }
}

根据内存的管理方不同，分为堆缓存区和直接缓存区，也就是Heap ByteBuf和Direct ByteBuf。

image.png

2.6 ByteBuf的自动释放

TailHandler自动释放
如果每个InboundHandler入站处理器，把最初的ByteBuf数据包一路往下传，那么TailHandler末尾处理器会自动释放掉入站的ByteBuf实例

（1）手动释放ByteBuf。具体的方式为调用byteBuf.release()。
（2）调用父类的入站方法将msg向后传递，依赖后面的处理器释放ByteBuf。具体的方式为调用基类的入站处理方法super.channelRead(ctx,msg)。

SimpleChannelInboundHandler自动释放

· 手动释放ByteBuf实例。
· 继承SimpleChannelInboundHandler，利用它的自动释放功能。