netty原理解析

Netty is an asynchronous event-driven network application framework for rapid development of maintainable high performance protocol servers & clients.

Netty是一个异步事件驱动的网络应用程序框架 用于快速开发可维护的高性能协议服务器和客户端。

使用场景

Dubbo RocketMQ Zookeeper Hadoop Spring-Cloud-Gateway 大数据、微服务、游戏、直播、IM、物联网等

netty的特性

• 不同类型传输层统一的API封装
• 基于灵活且可扩展的事件模型，可以清晰地分离关注点
• 异步非阻塞
• 高度可定制的线程模型
• 内置多种编解码功能，支持多种主流协议
• 高性能（高吞吐量、低延迟、更少的资源占用和不必要的内存copy）
• 没有额外的依赖，Netty 4.x JDK6 is enough
• 完整的SSL/TLS和StartTLS支持

功能模块

• Core核心模块 灵活且可扩展的事件模型 统一的传输层API 零拷贝的字节缓冲区
• Transport Services Sokect/Datagram (tcp/udp) HTTP Tunnel (Http隧道代理) In-VM Pipe (JVM进程内通信)
• Protocol Support Http/WebSocket/RTSP (应用层协议) SSL/StartTLS (传输层安全加密) Google Protocol Buffer (高效的数据存储格式) zlib/gizp Compression (数据压缩算法) Large File Transfer (大文件传输)

netty使用

Server示例

示例说明

ServerMain

• 创建bossGroup和workerGroup事件循环处理组
  • NioEventLoopGroup包含多个NioEventLoop
  • NioEventLoop对应一个线程循环，一个Selector，处理IO事件
  • bossGroup 处理接收连接请求，创建通道 (一个线程)
  • workerGroup 处理通道IO读写事件 (机器核心数 * 2)
• 创建serverHandler和loggingHandler事件处理器
• 通过ServerBootstrap进行server端配置，采用门面模式，封装内部细节，简化使用
  • 设置事件循环处理组 bossGroup & workerGroup
  • 设置通道实现类 NioServerSocketChannel
  • 设置通道配置项 ChannelOption.SO_BACKLOG (监听连接请求的队列大小)
  • 设置父通道事件处理器
  • 设置子通道事件处理器
    • ChannelPipeline channel管道，和Channel一一对应
    • ChannelPipeline 维护通道上行以及下行操作处理器链条
• 绑定指定端口 异步操作，返回ChannelFuture (可设置回调)
• 等待Channel关闭
• 优雅关闭事件循环处理组

EchoServerHandler

• @ChannelHandler.Sharable 声明该Handler无状态，可用于多个Channel的处理
• ChannelInboundHandlerAdapter 通道上行(被动)事件处理适配器，子类可以只实现自己关心的事件处理方法
• channelRead 从通道读取到数据时调用，将读取的数据写回到缓冲区 ctx.write
• channelReadComplete 从通道读取数据完毕时调用，将缓冲区的数据写到通道上 ctx.flush
• exceptionCaught I/O异常或者Handler内部处理异常调用，关闭通道 ctx.close()

Client示例

原理解析

Linux Socket系统调用

NIO相关概念

Selector 对应多个 SelectableChannel SelectableChannel 对应一个 Selector SelectionKey 与 SelectableChannel 一一对应

SelectionKey是一个用来记录SelectableChannel和Selector之间关系的对象，它由SelectableChannel的register()方法返回，并存储在Selector的多个集合中。它不仅记录了两个对象的引用，还包含了SelectableChannel感兴趣的操作，即OP_READ，OP_WRITE，OP_ACCEPT和OP_CONNECT。

server端启动流程

client端启动流程

Reactor线程模型

主要类图

ChannelPipeline与Handler

ChannelPipeline是ChannelHandler的编排管理容器，它内部维护了一个ChannelHandler的链表和迭代器，可以方便地实现ChannelHandler的查找、添加、替换和删除。消息经过ChannelPipeline在ChannelHandler中的处理流程

ChannelPipeline p = ...;
   p.addLast("1", new InboundHandlerA());
   p.addLast("2", new InboundHandlerB());
   p.addLast("3", new OutboundHandlerA());
   p.addLast("4", new OutboundHandlerB());
   p.addLast("5", new InboundOutboundHandlerX());
复制代码

inbound event order : 1, 2, 5 outbound action order : 5, 4, 3

ChannelHandler

• 无状态的ChannelHandler可以使用单例，即多个channel共享一个，减少对象创建，如LoggingHandler，用@ChannelHandler.Sharable标记
• 有状态的ChannelHandler，每个channel不能共享，各自拥有一个handler实例（大部分的业务handler都是有状态）

在实际业务场景中，ChannelPipeline通常需要添加如下几类ChannelHandler。

• 协议解码ChannelHandler
• 协议编码ChannelHandler
• 业务逻辑执行ChannelHandler

对于耗时的业务逻辑执行，例如访问数据库、中间件、第三方系统等，则需要切换到业务线程池中，避免阻塞Netty的NioEventLoop线程，导致消息无法接收和发送。

数据读流程

数据写流程

业务调用write方法后，经过ChannelPipeline职责链处理，消息被投递到发送缓冲区待发送，调用flush之后会执行真正的发送操作，底层通过调用Java NIO的SocketChannel进行非阻塞write操作，将消息发送到网络上

为了尽可能地提升性能，Netty采用了串行无锁化设计，在I/O线程内部进行串行操作，避免多线程竞争导致性能下降

ChannelOutboundBuffer

ChannelOutboundBuffer是Netty的发送缓冲队列，它基于链表来管理待发送的消息

发送次数限制当SocketChannel无法一次将所有待发送的ByteBuf/ByteBuffer写入网络时，需要决定是注册SelectionKey.OP_WRITE在下一次Selector轮询时继续发送，还是在当前位置循环发送，等到所有消息都发送完成再返回。

消息发送高低水位控制当消息队列中积压的待发送消息总字节数到达高水位时，修改Channel的状态为不可写，调用ChannelPipeline发送通知事件fireChannelWritabilityChanged，业务可以监听该事件及时获取链路可写状态

利用Netty的高低水位机制，可以防止在发送队列处于高水位时继续发送消息，导致积压更严重，甚至发生内存泄漏。在业务中合理利用Netty的高低水位机制，可以提升系统的可靠性

常用Handler

粘包/拆包

假设客户端向服务端连续发送了两个数据包，用packet1和packet2来表示，那么服务端收到的数据可以分为三种： 第一种情况，接收端正常收到两个数据包，即没有发生拆包和粘包的现象。

第二种情况，接收端只收到一个数据包，但是包含了发送端发送的两个数据包的信息，这种现象即为粘包。这种情况由于接收端不知道这两个数据包的界限，所以对于接收端来说很难处理。

第三种情况，这种情况有两种表现形式，如下图。接收端收到了两个数据包，但是这两个数据包要么是不完整的，要么就是多出来一块，这种情况即发生了拆包和粘包。这两种情况如果不加特殊处理，对于接收端同样是不好处理的。

发生TCP粘包、拆包主要是由于下面一些原因：

• 应用程序写入的数据大于套接字缓冲区大小，这将会发生拆包
• 应用程序写入数据小于套接字缓冲区大小，网卡将应用多次写入的数据发送到网络上，这将会发生粘包
• 进行MSS（最大报文长度）大小的TCP分段，当TCP报文长度-TCP头部长度>MSS的时候将发生拆包
• 接收方法不及时读取套接字缓冲区数据，这将发生粘包

ByteToMessageDecoder

FixedLengthFrameDecoder

@Override
protected final void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
    Object decoded = null;
    if (in.readableBytes() >= frameLength) {
        decoded = in.readRetainedSlice(frameLength);
    }
    if (decoded != null) {
        out.add(decoded);
    }
}
复制代码

DelimiterBasedFrameDecoder

使用定界符来分隔消息，比如/r/n

LengthFieldBasedFrameDecoder

使用消息中表示长度字段来定义消息的长度

WebSocket13FrameDecoder

WebSocket的协议格式

WebSocket的解析过程

计算framePayloadLength示例：

if (framePayloadLen1 == 126) {
    if (in.readableBytes() < 2) {
        return;
    }
    framePayloadLength = in.readUnsignedShort();
    if (framePayloadLength < 126) {
        protocolViolation(ctx, in, "invalid data frame length (not using minimal length encoding)");
        return;
    }
} else if (framePayloadLen1 == 127) {
    if (in.readableBytes() < 8) {
        return;
    }
    framePayloadLength = in.readLong();

    if (framePayloadLength < 65536) {
        protocolViolation(ctx, in, "invalid data frame length (not using minimal length encoding)");
        return;
    }
} else {
    framePayloadLength = framePayloadLen1;
}
复制代码

常见的使用方式

• pipeline.addLast(new XxxProtocolEncoder());
• pipeline.addLast(new XxxProtocolDecoder());
• pipeline.addLast(new BusinesHandler());

字节缓冲区 (数据容器)

ByteBuf

Netty提供了两个指针变量用于支持顺序读取read()和写入write()操作：readerIndex用于标识读取索引，writerIndex用于标识写入索引。两个位置指针将ByteBuf缓冲区划分为三部分.

顺序读写 read write调用ByteBuf的read操作时，从readerIndex处开始读取。从readerIndex到writerIndex的空间为可读的字节缓冲区；从writerIndex到capacity的空间为可写的字节缓冲区；从0到readerIndex的空间是已经读取的缓冲区，可以调用discard操作来重用这部分空间，以节约内存，防止ByteBuf不断扩张。

随机读写 get set可以随机指定读写的索引位置, ByteBuf会对其索引和长度等进行合法性校验，内部不会更改readerIndex和writerIndex，需要使用者自己控制。另外，set操作与write操作不同的是，它不支持动态扩展缓冲区，所以使用者必须保证当前的缓冲区可写的字节数大于需要写入的字节数，否则会抛出数组或者缓冲区越界异常。

注意事项

• ByteBuf的线程并发安全问题，特别要防止Netty NioEventLoop线程与应用线程并发操作ByteBuf。
• ByteBuf的申请和释放，避免忘记释放、重复释放，以及释放之后继续访问。(Netty某些场景下内部会隐式释放，当应用继续访问或者释放时就会发生异常)

引用计数

Netty4开始引入了引用计数的特性，缓冲区的生命周期可由引用计数管理，当缓冲区不再有用时，可快速返回给对象池或者分配器用于再次分配，从而大大提高性能，进而保证请求的实时处理。

引用计数两个基本规则

• 对象的初始引用计数为1
• 引用计数为0的对象不能再被使用，只能被释放

在代码中，可以使用retain()使引用计数增加1，使用release()使引用计数减少1，这两个方法都可以指定参数表示引用计数的增加值和减少值。当我们使用引用计数为0的对象时，将抛出异常。

谁负责释放对象通用的原则是：谁最后使用含有引用计数的对象，谁负责释放或销毁该对象。一般来说，有以下两种情况：

• 一个发送组件将引用计数对象传递给另一个接收组件时，发送组件无需负责释放对象，由接收组件决定是否释放。
• 一个消费组件消费了引用计数对象并且明确知道不再有其他组件使用该对象，那么该消费组件应该释放引用计数对象。

派生缓冲区通过duplicate()，slice()等等生成的派生缓冲区ByteBuf会共享原生缓冲区的内部存储区域。此外，派生缓冲区并没有自己独立的引用计数而需要共享原生缓冲区的引用计数。也就是说，当我们需要将派生缓冲区传入下一个组件时，一定要注意先调用retain()方法。

另外，实现ByteBufHolder接口的对象与派生缓冲区有类似的地方：共享所Hold缓冲区的引用计数，所以要注意对象的释放。在Netty，这样的对象包括DatagramPacket，HttpContent和WebSocketframe。

内存池

在API Gateway、RPC和流式处理框架中，请求和响应消息往往是“朝生熄灭”的，特别是频繁地申请和释放大块byte数组，会加重GC的负担及加大CPU资源占用率，通过内存池技术重用这些临时对象，可以降低GC频次和减少耗时，同时提升系统的吞吐量。

• GC回收或者引用队列回收效率不高，难以满足高性能的需求
• 缓冲区对象还需要尽可能的重用，避免创建大量短暂对象

Netty的内存池整体上参照jemalloc实现

PooledArena：代表内存中一大块连续的区域，PoolArena由多个Chunk组成，每个Chunk由多个Page组成。为了提升并发性能，内存池中包含一组PooledArena。

一个PoolArena内存块是由两个SubPagePools(用来存储零碎内存)和多个ChunkList组成，两个SubpagePools数组分别为tinySubpagePools和smallSubpagePools。每个ChunkList里包含多个Chunk按照双向链表排列，每个Chunk里包含多个Page（默认2048个），每个Page（默认大小为8k字节）由多个Subpage组成。

每个ChunkList里包含的Chunk数量会动态变化，比如当该chunk的内存利用率变化时会向其它ChunkList里移动。

内存分配的入口是PooledByteBufAllocator.buffer，返回PooledByteBuf，总体分为两步

• 第一：Recycler对象池中获取复用的ByteBuf对象
• 第二：为ByteBuf对象分配内存（从内存池获取）
  • 先从线程本地缓存PoolThreadCache申请
  • 再从全局内存PoolArena申请

PoolArena的分配过程

最终内存分配工作被委托给PoolArena（为了提高多线程并发，PoolArena是跟线程绑定的，不是一一对应关系）

• 对于小于PageSize大小的内存分配，会在tinySubPagePools和smallSubPagePools中分配，tinySubPagePools用来分配小于512字节的内存，smallSubPagePools用来分配大于512字节小于PageSize的内存；
• 对于大于PageSize小于ChunkSize的内存分配，会在PoolChunkList中的Chunk中分配
• 对于大于ChunkSize的内存分配，会之间直接创建非池化的Chunk来分配，并且该Chunk不会放在内存池中重用。

PoolChunk的分配过程

该二叉树将PoolChunk分11层, 第一层为1个16M, 第二层为2个8MB,第三层为4个4MB的内存块, 直到第11层为2048个8KB的内存块, 8kb的内存块称之为page。

PoolChunk用memoryMap和depthMap来表示二叉树，其中memoryMap存放的是PoolSubpage的分配信息，depthMap存放的是二叉树的深度。depthMap初始化之后就不再变化，而memoryMap则随着PoolSubpage的分配而改变。初始化时，memoryMap和depthMap的取值相同，都是节点所在层高。

节点的分配情况有如下三种可能：

1. memoryMap[id]=depthMap[id]：表示当前节点可分配内存
2. memoryMap[id]>depthMap[id]：表示当前节点至少一个子节点已经被分配，无法分配满足该深度的内存，但是可以分配更小一些的内存（通过空闲的子节点）
3. memoryMap[id]=最大深度（默认为11）+1：表示当前节点下的所有子节点都已经分配完，没有可用内存

分配一个内存大小为 chunkSize/2^k时，则先在第K层找第一个未使用的节点（从左到右），举例：

• 申请16M的内存, 那么直接在该二叉树第一层申请
• 申请32K的内存, 那么在该二叉树第8层申请
• 申请8K的内存, 那么在将直接在第11层申请

节点被标记为被占用之后，依次向上遍历更新父节点，直到根节点。将父节点的memoryMap[id]位置信息修改为两个子节点中的较小值

对于不足一个pagesize大小的内存分配，则由PoolSubpage进行处理，每一个PoolSubpage都会与PoolChunk里面的一个叶子节点映射起来, 然后将PoolSubpage根据用户申请的ElementSize化成几等分, 之后只要再次申请ElementSize大小的内存, 将直接从这个PoolSubpage中分配 (添加到PoolArena对应的SubPagePools中)。

PoolThreadCache

PoolThreadCache的内存来自于当前线程释放的

内存释放

当PooledByteBuf引用计数减为0时，会触发调用deallocate Recycler.recycle 归还 ByteBuf PoolArena.free 归还 内存 (byte[])

缓冲区泄露检测

由于JVM并没有意识到Netty实现的引用计数对象，它仍会将这些引用计数对象当做常规对象处理，也就意味着，当不为0的引用计数对象变得不可达时仍然会被GC自动回收。一旦被GC回收，引用计数对象将不再返回给创建它的对象池，这样便会造成内存泄露。

为了便于用户发现内存泄露，Netty提供了相应的检测机制并定义了四个检测级别：

• DISABLED 完全关闭内存泄露检测，并不建议
• SIMPLE 以1%的抽样率检测是否泄露，默认级别
• ADVANCED 抽样率同SIMPLE，但显示详细的泄露报告
• PARANOID 抽样率为100%，显示报告信息同ADVANCED

最佳实践

1. 单元测试和集成测试使用PARANOID级别
2. 使用SIMPLE级别运行足够长的时间以确定没有内存泄露，然后再将应用部署到集群
3. 如果发现有内存泄露，调到ADVANCED级别以提供寻找泄露的线索信息
4. 不要将有内存泄露的应用部署到整个集群

直接内存vs堆内存

堆内存是在JVM堆上分配的，由JVM负责进行垃圾回收 直接内存是直接在Native堆上分配，并不由JVM负责进行垃圾回收

创建和释放Direct Buffer的代价比Heap Buffer要高 Direct Buffer减少用户态和内核态之间的数据copy

对于涉及大量I/O的数据读写，建议使用Direct Buffer；而对于用于后端的业务消息编解码模块建议使用Heap Buffer

Zero-Copy

netty中的Zero-copy与传统意义的zero-copy不太一样,注意不要混淆:

• 传统的zero-copy是IO传输过程中，数据无需从内核态到用户态、用户态到内核态的数据拷贝，减少拷贝次数。
  • JAVA类库通过java.nio.channels.FileChannel. transferTo()方法支持zerocopy技术
• Netty的zero-copy则是在用户态，或者说传输层的zero-copy机制。

在协议传输过程中，通常需要拆包、合并包，常见的做法就是通过System.arrayCopy来复制需要的数据,但这样需要付出内容复制的开销. Netty通过ByteBuf.slice和Unpooled.wrappedBuffer等方法拆分、合并Buffer,做到无需拷贝数据。

异步Promise

JDK 的 Future 对象，该接口的方法如下：

Netty 扩展了 JDK 的 Future 接口，扩展的方法如下：

Future 对象有两种状态尚未完成和已完成，其中已完成又有三种状态：成功、失败、用户取消。

Future 接口中的方法都是 getter 方法而没有 setter 方法，也就是说这样实现的 Future 子类的状态是不可变的，如果我们想要变化，Netty 提供的解决方法是：使用可写的 Future 即 Promise。

Promise 接口继承自 Future 接口，它提供的 setter 方法与常见的 setter 方法大为不同。Promise 从 Uncompleted 到 Completed 的状态转变有且只能有一次，也就是说 setSuccess 和 setFailure 方法最多只会成功一个，此外，在 setSuccess 和 setFailure 方法中会通知注册到其上的监听者。

优雅退出

Runtime.getRuntime().addShutdownHook(new Thread() {
    @Override
    public void run() {
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
    }
});
复制代码

shutdownGracefully仅仅通过自旋CAS更改内部state NioEventLoop的run方法内部在执行一轮I/O操作以及Task之后，会判断state的值，如果需要关闭则执行closeAll和confirmShutdown方法，执行完毕则退出循环（即I/O线程)

closeAll处理逻辑

• 关闭所有的channel
• 清除outboundBuffer所有未发送的消息并释放资源
• 将SocketChannel从Selector上注销
• pipeline.fireChannelInactive
• pipeline.fireChannelUnregistered

confirmShutdown处理逻辑

• 取消所有的定时任务
• 执行队列中所有待执行的task
• 执行netty的ShutdownHook

最后NioEventLoop线程退出执行并关闭selector

性能调优

TCP建立连接/释放连接

操作系统TCP/IP参数调优

Netty性能调优

设置合理的线程数I/O工作线程数 默认值（CPU核数 * 2) 业务线程池

设置合理的心跳周期Netty通过IdleStateHandler来提供链路空闲检测机制

• 读空闲，链路持续时间T没有读取到任何消息
• 写空闲，链路持续时间T没有发送任何消息
• 读写空闲，链路持续时间T没有接收或者发送任何消息

合理使用内存池

防止I/O线程被意外阻塞

I/O线程和业务线程分离

并发连接数的兜底限制