1. 概述
1.1 Netty 是什么
Netty is an asynchronous event-driven network application framework for rapid development of maintainable high performance protocol servers & clients.
Netty 是一个异步的、基于事件驱动的网络应用框架,用于快速开发可维护、高性能的网络服务器和客户端
1.2 Netty 的地位
Netty 在 Java 网络应用框架中的地位就好比:Spring 框架在 JavaEE 开发中的地位
以下的框架都使用了 Netty,因为它们有网络通信需求!
- Cassandra - nosql 数据库
- Spark - 大数据分布式计算框架
- Hadoop - 大数据分布式存储框架
- RocketMQ - ali 开源的消息队列
- ElasticSearch - 搜索引擎
- GRPC - rpc 框架
- Dubbo - rpc 框架
- Spring 5.x - flux api 完全抛弃了 tomcat ,使用 Netty 作为服务器端
- Zookeeper - 分布式协调框架
1.3 Netty 的优势
需要自己构建协议
解决 TCP 传输问题,如粘包、半包
epoll 空轮询导致 CPU 100%
对 API 进行增强,使之更易用,如 FastThreadLocal => ThreadLocal,ByteBuf => ByteBuffer2. 入门案例
把 channel 理解为数据的通道
把 msg 理解为流动的数据,最开始输入是 ByteBuf,但经过 pipeline 的加工,会变成其它类型对象,最后输出又变成 ByteBuf
把 handler 理解为数据的处理工序
- 工序有多道,合在一起就是 pipeline,pipeline 负责发布事件(读、读取完成…)传播给每个 handler, handler 对自己感兴趣的事件进行处理(重写了相应事件处理方法)
- handler 分 Inbound 和 Outbound 两类
把 eventLoop 理解为处理数据的工人
- 工人可以管理多个 channel 的 io 操作,并且一旦工人负责了某个 channel,就要负责到底(绑定)
- 工人既可以执行 io 操作,也可以进行任务处理,每位工人有任务队列,队列里可以堆放多个 channel 的待处理任务,任务分为普通任务、定时任务
- 工人按照 pipeline 顺序,依次按照 handler 的规划(代码)处理数据,可以为每道工序指定不同的工人
public static void main(String[] args) {
new ServerBootstrap()
// 1. 创建 NioEventLoopGroup,可以简单理解为 线程池 + Selector
.group(new NioEventLoopGroup(), new NioEventLoopGroup())
// 2. 选择服务 socket 实现类,其中 NioServerSocketChannel 表示基于 NIO 的服务器端实现
.channel(NioServerSocketChannel.class)
// 3. 添加 handler 处理器
.childHandler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) throws Exception {
// 5. SocketChannel 的处理器,解码 ByteBuf => String
ch.pipeline().addLast(new StringDecoder());
// 6. SocketChannel 的自定义处理器,使用上一个处理器的处理结果
ch.pipeline().addLast(new SimpleChannelInboundHandler<String>() {
@Override
protected void channelRead0(ChannelHandlerContext channelHandlerContext, String msg) throws Exception {
log.info("收到消息 {}", msg);
}
});
}
})
// 4. ServerSocketChannel 绑定的监听端口
.bind(9527);
}
public static void main(String[] args) throws Exception {
ChannelFuture channelFuture = new Bootstrap()
// 1. 创建 NioEventLoopGroup,同 Server
.group(new NioEventLoopGroup())
// 2. 选择服务 socket 实现类,其中 NioServerSocketChannel 表示基于 NIO 的服务器端实现
.channel(NioSocketChannel.class)
// 3. 添加 handler 处理器
.handler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) throws Exception {
// 8. 消息会经过通道 handler 处理,这里是将 String => ByteBuf 发出
ch.pipeline().addLast(new StringEncoder());
}
})
// 4. 指定要连接的服务器和端口
.connect("127.0.0.1", 9527);
// 5. Netty 中很多方法都是异步的,如 connect,这时需要使用 sync 方法等待 connect 建立连接完毕
// 6. 获取 channel 对象,它即为通道抽象,可以进行数据读写操作
Channel channel = channelFuture.sync().channel();
// 7. 写入消息并清空缓冲区
channel.writeAndFlush(new Date() + ":hello world!");
}
3. 组件
3.1 EventLoopGroup
3.1.1 EventLoop
EventLoop 本质是一个单线程执行器(同时维护了一个 Selector),里面有 run 方法处理 Channel 上源源不断的 io 事件
继承关系
- 继承自
j.u.c.ScheduledExecutorService
因此包含了线程池中所有的方法 - 继承自 netty 自己的
OrderedEventExecutor
继承自 netty 自己的 EventExecutorGroup
- 实现了 Iterable 接口提供遍历 EventLoop 的能力
- 另有 next 方法获取集合中下一个 EventLoop
public static void main(String[] args) {
EventLoopGroup eventLoopGroup = new DefaultEventLoopGroup(2);
for (int i = 0; i < 5; i++) {
log.info("获取对象 {}", eventLoopGroup.next());
}
}
3.1.3 优雅关闭
优雅关闭shutdownGracefully()
方法。该方法会首先切换 EventLoopGroup 到关闭状态从而拒绝新的任务的加入,然后在任务队列的任务都处理完成后,停止线程的运行。从而确保整体应用是在正常有序的状态下退出的3.1.4 处理 IO 任务
channel 与 EventLoop 之间一一绑定,如果是自定义了线程池,也会与其一一绑定public static void main(String[] args) {
EventLoopGroup executors = new DefaultEventLoopGroup(2);
new ServerBootstrap()
// 1. 创建 NioEventLoopGroup,可以简单理解为 线程池 + Selector
.group(new NioEventLoopGroup(), new NioEventLoopGroup())
// 2. 选择服务 socket 实现类,其中 NioServerSocketChannel 表示基于 NIO 的服务器端实现
.channel(NioServerSocketChannel.class)
// 3. 添加 handler 处理器
.childHandler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) throws Exception {
ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
// 5. SocketChannel 的处理器,解码 ByteBuf => String
ch.pipeline().addLast(new StringDecoder());
// 6. 指定一个异步线程去处理
ch.pipeline().addLast(executors, "custom", new SimpleChannelInboundHandler<String>() {
@Override
protected void channelRead0(ChannelHandlerContext channelHandlerContext, String msg) throws Exception {
log.info("收到消息 {}", msg);
}
});
}
})
// 4. ServerSocketChannel 绑定的监听端口
.bind(9527);
}
3.1.5 处理普通任务
```java NioEventLoopGroup nioWorkers = new NioEventLoopGroup(2);
log.debug(“server start…”); Thread.sleep(2000); nioWorkers.execute(()->{ log.debug(“normal task…”); });
<a name="i3LCA"></a>
### 3.1.6 处理定时任务
```java
NioEventLoopGroup nioWorkers = new NioEventLoopGroup(2);
log.debug("server start...");
Thread.sleep(2000);
nioWorkers.scheduleAtFixedRate(() -> {
log.debug("running...");
}, 0, 1, TimeUnit.SECONDS);
3.2 Channel
3.2.1 channel API
close() 可以用来关闭 channel
closeFuture() 用来处理 channel 的关闭
sync() 方法作用是同步等待 channel 关闭
addListener() 方法是异步等待 channel 关闭
pipeline() 方法添加处理器
write() 方法将数据写入
writeAndFlush() 方法将数据写入并刷出
3.2.2 ChannelFuture
public static void main(String[] args) throws Exception {
ChannelFuture channelFuture = new Bootstrap()
// 1. 创建 NioEventLoopGroup,同 Server
.group(new NioEventLoopGroup())
// 2. 选择服务 socket 实现类,其中 NioServerSocketChannel 表示基于 NIO 的服务器端实现
.channel(NioSocketChannel.class)
// 3. 添加 handler 处理器
.handler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) throws Exception {
// 8. 消息会经过通道 handler 处理,这里是将 String => ByteBuf 发出
ch.pipeline().addLast(new StringEncoder());
}
})
// 4. 指定要连接的服务器和端口
.connect("127.0.0.1", 9527);
// 5. Netty 中很多方法都是异步的,如 connect,这时需要使用 sync 方法等待 connect 建立连接完毕
channelFuture.sync();
// 6. 获取 channel 对象,它即为通道抽象,可以进行数据读写操作
Channel channel = channelFuture.channel();
// 7. 写入消息并清空缓冲区
channel.writeAndFlush(new Date() + ":hello world!");
}
connect()
方法是异步的,意味着不等连接建立,方法执行就返回了。因此 channelFuture 对象中不能【立刻】获得到正确的 Channel 对象
sync()
方法是指同步等待连接建立完成
除了使用 sync()
方法,还可以使用监听器回调的方法
public static void main(String[] args) throws Exception {
ChannelFuture channelFuture = new Bootstrap()
// 1. 创建 NioEventLoopGroup,同 Server
.group(new NioEventLoopGroup())
// 2. 选择服务 socket 实现类,其中 NioServerSocketChannel 表示基于 NIO 的服务器端实现
.channel(NioSocketChannel.class)
// 3. 添加 handler 处理器
.handler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) throws Exception {
// 8. 消息会经过通道 handler 处理,这里是将 String => ByteBuf 发出
ch.pipeline().addLast(new StringEncoder());
}
})
// 4. 指定要连接的服务器和端口
.connect("127.0.0.1", 9527);
// 5. 使用监听器回调
channelFuture.addListener((ChannelFutureListener) future ->
// 6. 此时 channel 已经建立好,发送消息
future.channel().writeAndFlush(new Date() + ":hello world!")
);
}
3.2.3 CloseFuture
和上面的 ChannelFuture 一样,也有同步 sync()
和 异步 addListener()
两种方法
public static void main(String[] args) throws Exception {
NioEventLoopGroup executors = new NioEventLoopGroup();
ChannelFuture channelFuture = new Bootstrap()
// 1. 创建 NioEventLoopGroup,同 Server
.group(executors)
// 2. 选择服务 socket 实现类,其中 NioServerSocketChannel 表示基于 NIO 的服务器端实现
.channel(NioSocketChannel.class)
// 3. 添加 handler 处理器
.handler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) throws Exception {
// 8. 消息会经过通道 handler 处理,这里是将 String => ByteBuf 发出
ch.pipeline().addLast(new StringEncoder());
}
})
// 4. 指定要连接的服务器和端口
.connect("127.0.0.1", 9527);
// 5. Netty 中很多方法都是异步的,如 connect,这时需要使用 sync 方法等待 connect 建立连接完毕
channelFuture.sync();
// 6. 获取 channel 对象,它即为通道抽象,可以进行数据读写操作
Channel channel = channelFuture.channel();
// 7. 写入消息并清空缓冲区
channel.writeAndFlush(new Date() + ":hello world!");
// 8. 关闭 channel
channel.close();
// 同步等待关闭通道
// channel.closeFuture().sync();
// 异步等待通道关闭后的回调事件
channel.closeFuture().addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
log.info("通道已经关闭");
executors.shutdownGracefully();
}
});
}
3.3 Future & Promise
首先要说明 netty 中的 Future 与 jdk 中的 Future 同名,但是是两个接口,netty 的 Future 继承自 jdk 的 Future,而 Promise 又对 netty Future 进行了扩展
- jdk Future 只能同步等待任务结束(或成功、或失败)才能得到结果
- netty Future 可以同步等待任务结束得到结果,也可以异步方式得到结果,但都是要等任务结束
- netty Promise 不仅有 netty Future 的功能,而且脱离了任务独立存在,只作为两个线程间传递结果的容器 | 功能/名称 | jdk Future | netty Future | Promise | | —- | —- | —- | —- | | cancel | 取消任务 | - | - | | isCanceled | 任务是否取消 | - | - | | isDone | 任务是否完成,不能区分成功失败 | - | - | | get | 获取任务结果,阻塞等待 | - | - | | getNow | - | 获取任务结果,非阻塞,还未产生结果时返回 null | - | | await | - | 等待任务结束,如果任务失败,不会抛异常,而是通过 isSuccess 判断 | - | | sync | - | 等待任务结束,如果任务失败,抛出异常 | - | | isSuccess | - | 判断任务是否成功 | - | | cause | - | 获取失败信息,非阻塞,如果没有失败,返回null | - | | addLinstener | - | 添加回调,异步接收结果 | - | | setSuccess | - | - | 设置成功结果 | | setFailure | - | - | 设置失败结果 |
3.4 Handler & Pipeline
ChannelHandler 用来处理 Channel 上的各种事件,分为入站、出站两种。所有 ChannelHandler 被连成一串,就是 Pipeline
- 入站处理器通常是 ChannelInboundHandlerAdapter 的子类,主要用来读取客户端数据,写回结果
- 出站处理器通常是 ChannelOutboundHandlerAdapter 的子类,主要对写回结果进行加工
打个比喻,每个 Channel 是一个产品的加工车间,Pipeline 是车间中的流水线,ChannelHandler 就是流水线上的各道工序,而后面要讲的 ByteBuf 是原材料,经过很多工序的加工:先经过一道道入站工序,再经过一道道出站工序最终变成产品
public static void main(String[] args) {
new ServerBootstrap()
.group(new NioEventLoopGroup(), new NioEventLoopGroup())
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) throws Exception {
ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("inbound-1");
super.channelRead(ctx, msg);
}
});
ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("inbound-2");
super.channelRead(ctx, msg);
}
});
ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("inbound-3");
// ① 响应客户端
ctx.channel().writeAndFlush(msg);
}
});
ch.pipeline().addLast(new ChannelOutboundHandlerAdapter() {
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.info("outbound-4");
super.write(ctx, msg, promise);
}
});
ch.pipeline().addLast(new ChannelOutboundHandlerAdapter() {
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.info("outbound-5");
super.write(ctx, msg, promise);
}
});
ch.pipeline().addLast(new ChannelOutboundHandlerAdapter() {
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.info("outbound-6");
// ②
super.write(ctx, msg, promise);
}
});
}
})
// 4. ServerSocketChannel 绑定的监听端口
.bind(9527);
}
inbound-1 -> inbound-2 -> inbound-3 -> outbound-6 -> outbound-5 -> outbound-4
可以看到,ChannelInboundHandlerAdapter 是按照 addLast 的顺序执行的,而 ChannelOutboundHandlerAdapter 是按照 addLast 的逆序执行的。ChannelPipeline 的实现是一个 ChannelHandlerContext(包装了 ChannelHandler) 组成的双向链表
在入站处理器中,ctx.fireChannelRead(msg)
是 调用下一个入站处理器,ctx.channel().write(msg)
会 从尾部开始触发 后续出站处理器的执行
在出站处理器中,ctx.write(msg, promise)
的调用也会 触发上一个出站处理器
ctx.channel().write(msg)
VS ctx.write(msg)
- 都会触发出站处理器的执行
ctx.channel().write(msg)
从尾部开始查找出站处理器ctx.write(msg)
是从当前节点找上一个出站处理器- 如果 ① 处的
ctx.channel().write(msg)
改为ctx.write(msg)
仅会打印 1 2 3,因为节点 3 之前没有其它出站处理器了 - 如果 ② 处的
ctx.write(msg, promise)
如果改为ctx.channel().write(msg)
会打印 1 2 3 6 6 6…(死循环) 因为ctx.channel().write()
是从尾部开始查找,结果又是节点6 自己
3.5 ByteBuf
3.5.1 创建
创建了一个默认的 ByteBuf(池化基于直接内存的 ByteBuf),初始容量是 10
ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(10);
3.5.2 直接内存和堆内存
可以使用下面的代码来创建池化基于堆的 ByteBuf
ByteBuf buffer = ByteBufAllocator.DEFAULT.heapBuffer(10);
也可以使用下面的代码来创建池化基于直接内存的 ByteBuf
ByteBuf buffer = ByteBufAllocator.DEFAULT.directBuffer(10);
直接内存创建和销毁的代价昂贵,但读写性能高(少一次内存复制),适合配合池化功能一起用
直接内存对 GC 压力小,因为这部分内存不受 JVM 垃圾回收的管理,但也要注意及时主动释放
3.5.3 池化和非池化
池化的最大意义在于可以重用 ByteBuf,优点有
- 没有池化,则每次都得创建新的 ByteBuf 实例,这个操作对直接内存代价昂贵,就算是堆内存,也会增加 GC 压力
- 有了池化,则可以重用池中 ByteBuf 实例,并且采用了与 jemalloc 类似的内存分配算法提升分配效率
- 高并发时,池化功能更节约内存,减少内存溢出的可能
池化功能是否开启,可以通过下面的系统环境变量来设置
-Dio.netty.allocator.type={unpooled|pooled}
4.1 以后,非 Android 平台默认启用池化实现,Android 平台启用非池化实现
4.1 之前,池化功能还不成熟,默认是非池化实现
3.5.4 组成
3.5.5 写入
这些方法的未指明返回值的,其返回值都是 ByteBuf,意味着可以链式调用 网络传输,默认习惯是 Big Endian
方法签名 | 含义 | 备注 |
---|---|---|
writeBoolean(boolean value) | 写入 boolean 值 | 用一字节 01|00 代表 true|false |
writeByte(int value) | 写入 byte 值 | |
writeShort(int value) | 写入 short 值 | |
writeInt(int value) | 写入 int 值 | Big Endian,即 0x250,写入后 00 00 02 50 |
writeIntLE(int value) | 写入 int 值 | Little Endian,即 0x250,写入后 50 02 00 00 |
writeLong(long value) | 写入 long 值 | |
writeChar(int value) | 写入 char 值 | |
writeFloat(float value) | 写入 float 值 | |
writeDouble(double value) | 写入 double 值 | |
writeBytes(ByteBuf src) | 写入 netty 的 ByteBuf | |
writeBytes(byte[] src) | 写入 byte[] | |
writeBytes(ByteBuffer src) | 写入 nio 的 ByteBuffer | |
int writeCharSequence(CharSequence sequence, Charset charset) | 写入字符串 |
3.5.6 扩容
如何写入后数据大小未超过 512,则选择下一个 16 的整数倍,例如写入后大小为 12 ,则扩容后 capacity 是 16
如果写入后数据大小超过 512,则选择下一个 2^n,例如写入后大小为 513,则扩容后 capacity 是 2^10=1024(2^9=512 已经不够了)
3.5.7 读取
buffer.readByte()
读过的内容,就属于废弃部分了,再读只能读那些尚未读取的部分
如果需要重复读取,可以在 read 前先做个标记 mark,这时要重复读取的话,重置到标记位置 reset
buffer.markReaderIndex();
System.out.println(buffer.readInt());
log(buffer);
buffer.resetReaderIndex();
log(buffer);
3.5.8 retain 和 release
由于 Netty 中有堆外内存的 ByteBuf 实现,堆外内存最好是手动来释放,而不是等 GC 垃圾回收
- UnpooledHeapByteBuf 使用的是 JVM 内存,只需等 GC 回收内存即可
- UnpooledDirectByteBuf 使用的就是直接内存了,需要特殊的方法来回收内存
- PooledByteBuf 和它的子类使用了池化机制,需要更复杂的规则来回收内存
Netty 这里采用了引用计数法来控制回收内存,每个 ByteBuf 都实现了 ReferenceCounted 接口
- 每个 ByteBuf 对象的初始计数为 1
- 调用 release 方法计数减 1,如果计数为 0,ByteBuf 内存被回收
- 调用 retain 方法计数加 1,表示调用者没用完之前,其它 handler 即使调用了 release 也不会造成回收
- 当计数为 0 时,底层内存会被回收,这时即使 ByteBuf 对象还在,其各个方法均无法正常使用
基本规则是,谁是最后使用者,谁负责 release
- 起点,对于 NIO 实现来讲,在
io.netty.channel.nio.AbstractNioByteChannel.NioByteUnsafe#read()
方法中首次创建 ByteBuf 放入 pipeline - 入站 ByteBuf 处理原则
- 对原始 ByteBuf 不做处理,调用
ctx.fireChannelRead(msg)
向后传递,这时无须 release - 将原始 ByteBuf 转换为其它类型的 Java 对象,这时 ByteBuf 就没用了,必须 release
- 如果不调用
ctx.fireChannelRead(msg)
向后传递,那么也必须 release - 注意各种异常,如果 ByteBuf 没有成功传递到下一个 ChannelHandler,必须 release
- 假设消息一直向后传,那么 TailContext 会负责释放未处理消息(原始的 ByteBuf)
- 对原始 ByteBuf 不做处理,调用
- 出站 ByteBuf 处理原则
- 出站消息最终都会转为 ByteBuf 输出,一直向前传,由 HeadContext flush 后 release
- 异常处理原则
- 有时候不清楚 ByteBuf 被引用了多少次,但又必须彻底释放,可以循环调用 release 直到返回 true
```java
// io.netty.channel.DefaultChannelPipeline#onUnhandledInboundMessage(java.lang.Object)
protected void onUnhandledInboundMessage(Object msg) {
try {
logger.debug(
} finally { ReferenceCountUtil.release(msg); } }"Discarded inbound message {} that reached at the tail of the pipeline. " +
"Please check your pipeline configuration.", msg);
- 有时候不清楚 ByteBuf 被引用了多少次,但又必须彻底释放,可以循环调用 release 直到返回 true
```java
// io.netty.channel.DefaultChannelPipeline#onUnhandledInboundMessage(java.lang.Object)
protected void onUnhandledInboundMessage(Object msg) {
try {
logger.debug(
// io.netty.util.ReferenceCountUtil#release(java.lang.Object) public static boolean release(Object msg) { if (msg instanceof ReferenceCounted) { return ((ReferenceCounted) msg).release(); } return false; } ```
3.5.9 slice
【零拷贝】的体现之一,对原始 ByteBuf 进行切片成多个 ByteBuf,切片后的 ByteBuf 并没有发生内存复制,还是使用原始 ByteBuf 的内存,切片后的 ByteBuf 维护独立的 read,write 指针
3.5.10 duplicate
【零拷贝】的体现之一,就好比截取了原始 ByteBuf 所有内容,并且没有 max capacity 的限制,也是与原始 ByteBuf 使用同一块底层内存,只是读写指针是独立的
3.5.11 copy
会将底层内存数据进行深拷贝,因此无论读写,都与原始 ByteBuf 无关
3.5.12 CompositeByteBuf
【零拷贝】的体现之一,可以将多个 ByteBuf 合并为一个逻辑上的 ByteBuf,避免拷贝
3.5.13 Unpooled
Unpooled 是一个工具类,类如其名,提供了非池化的 ByteBuf 创建、组合、复制等操作
3.5.14 ByteBuf 优势
池化 - 可以重用池中 ByteBuf 实例,更节约内存,减少内存溢出的可能
读写指针分离,不需要像 ByteBuffer 一样切换读写模式
可以自动扩容
很多地方体现零拷贝,例如 slice、duplicate、CompositeByteBuf