知识点前文请阅读：网络编程-Netty篇
概述
-
Netty 是什么？
Netty is an asynchronous event-driven network application framework
for rapid development of maintainable high performance protocol servers & clients.
Netty 是一个异步的、基于事件驱动的网络应用框架，用于快速开发可维护、高性能的网络服务器和客户端
-
Netty 的作者
Trustin Lee。他还是Mina框架的重要贡献者
-
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 – 分布式协调框架
-
Netty 的优势
Netty vs NIO，工作量大，bug 多
需要自己构建协议
解决 TCP 传输问题，如粘包、半包
epoll 空轮询导致 CPU 100%
对 API 进行增强，使之更易用
Netty vs 其它网络应用框架
Mina 由 apache 维护，将来 3.x 版本可能会有较大重构，破坏 API 向下兼容性，Netty 的开发迭代更迅速，API 更简洁、文档更优秀
久经考验，16年，Netty 版本
2.x 2004
3.x 2008
4.x 2013
5.x 已废弃（没有明显的性能提升，维护成本高）
Hello World
- 开发一个简单的服务器端和客户端
客户端向服务器端发送 hello, world
服务器仅接收，不返回
添加依赖
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.72.Final</version>
</dependency>服务器端
public static void main(String[] args) {
// 1、ServerBootstrap : 服务器端的启动器，负责组装netty组件
new ServerBootstrap()
// 2、 BossEventLoop WorkerEventLoop，一个selector和一个thread就叫 EventLoop
// 表示加入一组 EventLoop（selector和thread）
.group(new NioEventLoopGroup())
// 3、选择一个Channel的实现，netty包括很多的io实现，例如OIO、BIO、Epoll，这里使用NIO的实现
// .option(ChannelOption.SO_RCVBUF, 10)// 设置服务器端接收缓冲区大小，单位字节
.channel(NioServerSocketChannel.class)
// 4、boss 处理连接的，worker 处理读写，这里的 child 职责和 worker 的职责一样，决定了具体的处理操作（handler）
.childHandler(
// 5、表示 和 客户端连接后，进行数据读写的通道（NIO示例中的 SocketChannel）
// Initializer 初始化器，负责添加其他的handler
new ChannelInitializer<NioSocketChannel>() {
@Override // 在连接建立后 处理 accept(由netty内部代码完成)、read 事件
protected void initChannel(NioSocketChannel channel) throws Exception {
// 6、添加具体的 handler
channel.pipeline().addLast(new StringDecoder());// decoder解码器，因为数据传输过来，都是通过字节(ByteBuf)进行传输，服务器端这边需要使用 StringDecoder 将字节转为字符串，客户端需要使用 StringEncoder 编码器将字符串转成字节进行传输
channel.pipeline().addLast(new ChannelInboundHandlerAdapter() { // 自定义的 handler
@Override // 处理 read 读事件
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.err.println(msg);
}
});
}
})
// 7、服务器使用的端口
.bind(8080);
}客户端
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 channel) throws Exception {
// 7、发送数据之前会走到这里，将数据 ‘hello, world’ 转成 字节ByteBuf
channel.pipeline().addLast(new StringEncoder()); // 客服端要向服务器端发送数据，数据是字节类型的，所以要 添加 StringEncoder 处理器，把 字符串转成字节
}
})
// 5、连接服务器，connect方法异步非阻塞
.connect(new InetSocketAddress("localhost", 8080))
.sync() // 阻塞方法，直到连接建立后唤醒线程
.channel() // 表示的连接到服务端的SocketChannel
// 6、向服务器端发送数据
.writeAndFlush("hello, world");
}流程梳理
一开始需要树立正确的观念
把 channel 理解为数据的通道
把 msg 理解为流动的数据，最开始输入是 ByteBuf，但经过 pipeline 的加工，会变成其它类型对象，最后输出又变成 ByteBuf
把 handler 理解为数据的处理工序
工序有多道，合在一起就是 pipeline，pipeline 负责发布事件（读、读取完成…）传播给每个 handler， handler 对自己感兴趣的事件进行处理（重写了相应事件处理方法）
handler 分 Inbound 和 Outbound 两类
把 eventLoop 理解为处理数据的工人
工人可以管理多个 channel 的 io 操作，并且一旦工人负责了某个 channel，就要负责到底（绑定）
工人既可以执行 io 操作，也可以进行任务处理，每位工人有任务队列，队列里可以堆放多个 channel 的待处理任务，任务分为普通任务、定时任务
工人按照 pipeline 顺序，依次按照 handler 的规划（代码）处理数据，可以为每道工序指定不同的工人
组件
-
EventLoop
EventLoop 本质是一个单线程执行器（同时维护了一个 Selector），里面有 run 方法处理 Channel 上源源不断的 io 事件。
它的继承关系比较复杂
一条线是继承自 j.u.c.ScheduledExecutorService 因此包含了线程池中所有的方法
另一条线是继承自 netty 自己的 OrderedEventExecutor，
提供了 boolean inEventLoop(Thread thread) 方法判断一个线程是否属于此 EventLoop
提供了 parent 方法来看看自己属于哪个 EventLoopGroup
EventLoopGroup 是一组 EventLoop，Channel 一般会调用 EventLoopGroup 的 register 方法来绑定其中一个 EventLoop，后续这个 Channel 上的 io 事件都由此 EventLoop 来处理（保证了 io 事件处理时的线程安全）
继承自 netty 自己的 EventExecutorGroup
实现了 Iterable 接口提供遍历 EventLoop 的能力
另有 next 方法获取集合中下一个 EventLoop
NioEventLoopGroup // 处理 NIO、普通任务、定时任务
DefaultEventLoopGroup // 处理 普通任务、定时任务
- 细化职责一
上面的代码示例中，服务器端的group只传入了一个 EventLoopGroup，netty推荐EventLoop细分自己的相应职责，所以对group代码改进如下
// boss：老板，只负责接待客户（accept），构造方法指定线程数量
NioEventLoopGroup bossEventLoopGroup = new NioEventLoopGroup();
// work: 工人，负责客户的需求（read）, 构造方法指定线程数量，不指定的情况下默认使用cpu核数*2的线程数量
NioEventLoopGroup workerEventLoopGroup = new NioEventLoopGroup(2);
new ServerBootstrap()
.group(bossEventLoopGroup, workerEventLoopGroup)
......- 细化职责二
一个WorkerEventLoop可以管理多个channle(一个工人可以做多种事情)。假如现在WorkerEventLoop管理了1000个channle，其中一个channle的耗时5秒，势必会影响剩下的channle的操作。
优化：WorkerEventLoop用来处理耗时短的操作，我们再招2个工人处理耗时长的操作，我们知道 NioEventLoopGroup 是可以处理 nio、普通任务、定时任务的，对于NIO操作已经有WorkerEventLoop（NioEventLoopGroup ）来做了，耗时的任务只是在服务端这里，如果没有IO交互，所以招收两个做普通任务的工人即可(当然招收NioEventLoopGroup也可)。代码如下
public static void main(String[] args) {
// boss：老板，只负责接待客户（accept），构造方法指定线程数量
NioEventLoopGroup bossEventLoopGroup = new NioEventLoopGroup();
// work: 工人，负责客户的需求（read）, 构造方法指定线程数量，不指定的情况下默认使用cpu核数*2的线程数量
NioEventLoopGroup workerEventLoopGroup = new NioEventLoopGroup(2);
// 普通EventLoopGroup
EventLoopGroup generalWorkerEventLoopGroup = new DefaultEventLoopGroup(2);
new ServerBootstrap()
.group(bossEventLoopGroup, workerEventLoopGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel channel) throws Exception {
channel.pipeline().addLast(new ChannelInboundHandlerAdapter() { // 第一个 handler
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf byteBuf = (ByteBuf) msg;
log.debug(byteBuf.toString(StandardCharsets.UTF_8));
// 本 handler 操作处理完后，要将消息交给下一个 handler
ctx.fireChannelRead(msg);
}
}).addLast(generalWorkerEventLoopGroup, "generalWorker", new ChannelInboundHandlerAdapter() {
// 第二个 耗时handler, 第一个参数 指定 EventLoopGroup，第二个参数 给 handler 自定义一个名称，第三个参数 具体执行的 handler
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
// 这里是耗时操作
ByteBuf byteBuf = (ByteBuf) msg;
log.debug(byteBuf.toString(StandardCharsets.UTF_8));
}
});
}
})
.bind(8080);
}-
channel
channel 的主要作用
close() 可以用来关闭 channel，异步
closeFuture() 用来处理 channel 的关闭
sync 方法作用是同步等待 channel 关闭
而 addListener 方法是异步等待 channel 关闭
pipeline() 方法添加处理器
write() 方法将数据写入，等缓冲区满了或者调用flush方法才发送数据
flush() 刷出缓冲区数据
writeAndFlush() 方法将数据写入并刷出
- ChannelFuture
sync 和 addListener 区别
- channelFuture.sync() // 主线程 同步等待 连接成功后处理操作
- channelFuture.addListener() // 让nio线程异步处理连接成功后的操作
示例：
方式一：sync
public static void main(String[] args) throws InterruptedException {
// 2.带有future、promise 的类型都是和异步方法配套使用，用来处理结果
ChannelFuture channelFuture = new Bootstrap()
.group(new NioEventLoopGroup())
.channel(NioSocketChannel.class)
.handler(
new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel channel) throws Exception {
channel.pipeline().addLast(new StringEncoder());
}
})
// 1.连接到服务器
// 异步非阻塞，main 发起调用，连接的线程是 nio 线程
.connect(new InetSocketAddress("localhost", 8080));
// 方式一
// 2.1 使用 sync 方法同步 connect 的连接结果
channelFuture.sync(); // 阻塞住当前主线程，直到等待 nio 线程 连接到服务器完成
// 不加 sync 方法，线程没有阻塞，直接来拿channel，但是客户端和服务端的连接还未建立成功，拿到的 channel 不是正确的，导致channel.writeAndFlush的消息发送不成功
Channel channel = channelFuture.channel();
log.debug(" ================ {}", channel);
channel.writeAndFlush("hello");
System.err.println("");
}方式二：addListener
public static void main(String[] args) throws InterruptedException {
// 2.带有future、promise 的类型都是和异步方法配套使用，用来处理结果
ChannelFuture channelFuture = new Bootstrap()
.group(new NioEventLoopGroup())
.channel(NioSocketChannel.class)
.handler(
new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel channel) throws Exception {
channel.pipeline().addLast(new StringEncoder());
}
})
// 1.连接到服务器
// 异步非阻塞，main 发起调用，连接的线程是 nio 线程
.connect(new InetSocketAddress("localhost", 8080));
// 方式二
// 2.2 使用 addListener(回调对象) 方法异步处理结果
// 主线程不等待线程的连接了，nio线程进行连接，nio连接到服务端成功后，告诉主线程说：“我nio已经连接成功了，你把要做什么事情告诉我”
channelFuture.addListener(new ChannelFutureListener() {
@Override // nio连接成功后，nio 线程需要做的操作
public void operationComplete(ChannelFuture channelFuture) throws Exception {
Channel channel = channelFuture.channel();
log.debug(" ================ {}", channel);
channel.writeAndFlush("hello");
System.err.println("");
}
});
}-
如何正确处理 客户端传输通道 channel 关闭以后的操作?
示例：
方式一：sync 同步等待 channel 关闭，阻塞主线程
public static void main(String[] args) throws InterruptedException {
ChannelFuture channelFuture = new Bootstrap().group(new NioEventLoopGroup())
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) throws Exception {
ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
ch.pipeline().addLast(new StringEncoder());
}
}).connect(new InetSocketAddress("localhost", 8080));
Channel channel = channelFuture.sync().channel();
log.debug("连接成功");
new Thread(() ->{
// 接收控制台输入，输入q时结束，输入其他内容就发送给服务器端
Scanner scanner = new Scanner(System.in);
while (true) {
String line = scanner.nextLine();
if ("q".equals(line)) {
channel.close(); // 关闭 channel
log.debug("处理关闭之后的操作-错误的处理位置");
break;
}
channel.writeAndFlush(line);
}
}, "input").start();
// 获取 CloseFuture 对象，处理 channel 关闭后的操作
ChannelFuture closeFuture = channel.closeFuture();
// 方式一：调用 sync 方法同步等待channel 关闭，阻塞主线程
log.debug("等待 channel 关闭...");
closeFuture.sync();
log.debug("处理关闭之后的操作");
}打印结果 （使用netty提供的调试工具LoggingHandler，可以看到channel内部的操作步骤）
11:48:10.840 [nioEventLoopGroup-2-1] DEBUG io.netty.handler.logging.LoggingHandler - [id: 0xaa736fa6] REGISTERED
11:48:10.840 [nioEventLoopGroup-2-1] DEBUG io.netty.handler.logging.LoggingHandler - [id: 0xaa736fa6] CONNECT: localhost/127.0.0.1:8080
11:48:10.844 [nioEventLoopGroup-2-1] DEBUG io.netty.handler.logging.LoggingHandler - [id: 0xaa736fa6, L:/127.0.0.1:63033 - R:localhost/127.0.0.1:8080] ACTIVE
11:48:10.845 [main] DEBUG com.lixx.demo.netty.CloseFutureClient - 连接成功
11:48:10.846 [main] DEBUG com.lixx.demo.netty.CloseFutureClient - 等待 channel 关闭...
q
11:48:15.099 [input] DEBUG com.lixx.demo.netty.CloseFutureClient - 处理关闭之后的操作-错误的处理位置 // 可以看到这里处理关闭之后的操作，下一行输出可以看到是在 channel关闭之前
11:48:15.099 [nioEventLoopGroup-2-1] DEBUG io.netty.handler.logging.LoggingHandler - [id: 0xaa736fa6, L:/127.0.0.1:63033 - R:localhost/127.0.0.1:8080] CLOSE
11:48:15.101 [main] DEBUG com.lixx.demo.netty.CloseFutureClient - 处理关闭之后的操作
11:48:15.101 [nioEventLoopGroup-2-1] DEBUG io.netty.handler.logging.LoggingHandler - [id: 0xaa736fa6, L:/127.0.0.1:63033 ! R:localhost/127.0.0.1:8080] INACTIVE
11:48:15.101 [nioEventLoopGroup-2-1] DEBUG io.netty.handler.logging.LoggingHandler - [id: 0xaa736fa6, L:/127.0.0.1:63033 ! R:localhost/127.0.0.1:8080] UNREGISTERED方式二：addListener(回调对象) 方法异步处理关闭后的操作
public static void main(String[] args) throws InterruptedException {
ChannelFuture channelFuture = new Bootstrap().group(new NioEventLoopGroup())
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) throws Exception {
ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
ch.pipeline().addLast(new StringEncoder());
}
}).connect(new InetSocketAddress("localhost", 8080));
Channel channel = channelFuture.sync().channel();
log.debug("连接成功");
new Thread(() ->{
// 接收控制台输入，输入q时结束，输入其他内容就发送给服务器端
Scanner scanner = new Scanner(System.in);
while (true) {
String line = scanner.nextLine();
if ("q".equals(line)) {
channel.close(); // 关闭 channel
log.debug("处理关闭之后的操作-错误的处理位置");
break;
}
channel.writeAndFlush(line);
}
}, "input").start();
// 获取 CloseFuture 对象，处理 channel 关闭后的操作
ChannelFuture closeFuture = channel.closeFuture();
// 方式二：使用 addListener(回调对象) 方法异步处理关闭后的操作
closeFuture.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture channelFuture) throws Exception {
log.debug("处理关闭之后的操作");
}
});
}打印结果
11:54:04.559 [nioEventLoopGroup-2-1] DEBUG io.netty.handler.logging.LoggingHandler - [id: 0x5771d47f] REGISTERED
11:54:04.559 [nioEventLoopGroup-2-1] DEBUG io.netty.handler.logging.LoggingHandler - [id: 0x5771d47f] CONNECT: localhost/127.0.0.1:8080
11:54:04.563 [nioEventLoopGroup-2-1] DEBUG io.netty.handler.logging.LoggingHandler - [id: 0x5771d47f, L:/127.0.0.1:63164 - R:localhost/127.0.0.1:8080] ACTIVE
11:54:04.566 [main] DEBUG com.lixx.demo.netty.CloseFutureClient - 连接成功
q
11:54:14.540 [input] DEBUG com.lixx.demo.netty.CloseFutureClient - 处理关闭之后的操作-错误的处理位置
11:54:14.540 [nioEventLoopGroup-2-1] DEBUG io.netty.handler.logging.LoggingHandler - [id: 0x5771d47f, L:/127.0.0.1:63164 - R:localhost/127.0.0.1:8080] CLOSE
11:54:14.541 [nioEventLoopGroup-2-1] DEBUG com.lixx.demo.netty.CloseFutureClient - 处理关闭之后的操作
11:54:14.542 [nioEventLoopGroup-2-1] DEBUG io.netty.handler.logging.LoggingHandler - [id: 0x5771d47f, L:/127.0.0.1:63164 ! R:localhost/127.0.0.1:8080] INACTIVE
11:54:14.542 [nioEventLoopGroup-2-1] DEBUG io.netty.handler.logging.LoggingHandler - [id: 0x5771d47f, L:/127.0.0.1:63164 ! R:localhost/127.0.0.1:8080] UNREGISTERED-
优雅的结束进程
当客户端传输通道关闭后，客户端进程并没有结束，因为 group 中还有线程在运行
public static void main(String[] args) throws InterruptedException {
NioEventLoopGroup group = new NioEventLoopGroup();
ChannelFuture channelFuture = new Bootstrap()
.group(group)
......
// 获取同步连接服务器端成功的 channel
Channel channel = channelFuture.sync().channel();
// 获取 closeFuture 对象
ChannelFuture closeFuture = channel.closeFuture();
// 关闭 channel
channel.close();
// 同步等待 channel 关闭完成
closeFuture.sync();
// 优雅的结束客户端进程
group.shutdownGracefully();
}思考：netty中的异步，使用那么多线程切换的执行事件的好处是什么？
答：提高吞吐量
单线程没法异步提高效率，必须配合多线程、多核 cpu 才能发挥异步的优势
异步并没有缩短每个请求到响应的时间，反而有所增加(线程切换)，netty 提高的是吞吐量，即单位时间内处理的请求数量
合理进行任务拆分，也是利用异步的关键
-
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 | – | – | 设置失败结果 |
示例：
jdk Future
public static void main(String[] args) throws Exception {
// 1.线程池
ExecutorService service = Executors.newFixedThreadPool(2);
// 2.提交任务
Future<Object> future = service.submit(new Callable<Object>() {
@Override // 执行事件逻辑的是线程池中的一个线程
public Object call() throws Exception {
// 假如处理事件逻辑 耗费了 3秒
log.debug("处理事件逻辑");
Thread.sleep(3000L);
// 返回处理结果
return 100;
}
});
// 3.主线程需要 事件处理的结果
log.debug("等待事件处理结果。。。");
Object o = future.get();// 阻塞方法
log.debug("拿到事件处理结果：{}", o);
}netty Future
public static void main(String[] args) throws Exception {
// 1.线程池 EventLoopGroup 中可以有多个线程，EventLoop 中只有一个线程
NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
// 拿到一个线程
EventLoop eventLoop = eventLoopGroup.next();
// 2.提交任务
Future<Object> future = eventLoop.submit(new Callable<Object>() {
@Override
public Object call() throws Exception {
// 假如处理事件逻辑 耗费了 3秒
log.debug("处理事件逻辑");
Thread.sleep(3000L);
// 返回处理结果
return 200;
}
});
// 3.主线程需要 事件处理的结果
// 同步方式，和 jdk future的效果一样
// Object o = future.get();
future.addListener(new GenericFutureListener<Future<? super Object>>() {
@Override
public void operationComplete(Future<? super Object> future) throws Exception {
log.debug("拿到事件处理结果：{}", future.getNow());
}
});
}netty Promise
public static void main(String[] args) throws Exception {
// 1. 拿到一个线程
EventLoop eventLoop = new NioEventLoopGroup().next();
// 2. 创建结果容器
DefaultPromise<Integer> promise = new DefaultPromise<>(eventLoop);
new Thread(() -> {
// 3. 任意一个线程执行事件，执行完成后可以向 promise 容器中填充结果
log.debug("开始处理事件逻辑");
try {
Thread.sleep(3000L);
// 填充结果
promise.setSuccess(300);
} catch (InterruptedException e) {
e.printStackTrace();
promise.setFailure(e.getCause());
}
}).start();
// 4. 接收结果
log.debug("等待事件处理结果。。。");
log.debug("拿到事件处理结果：{}", promise.get());
}-
Handler & Pipeline
ChannelHandler 用来处理 Channel 上的各种事件，分为入站、出站两种。所有 ChannelHandler 被连成一串，就是 Pipeline
入站处理器通常是 ChannelInboundHandlerAdapter 的子类，主要用来读取客户端数据，写回结果
出站处理器通常是 ChannelOutboundHandlerAdapter 的子类，主要对写回结果进行加工
打个比喻，每个 Channel 是一个产品的加工车间，Pipeline 是车间中的流水线，ChannelHandler 就是流水线上的各道工序，ByteBuf 是原材料，经过很多工序的加工：先经过一道道入站工序，再经过一道道出站工序最终变成产品
pipeline 的执行顺序：入站 In 是按照加入 pipeline 的正序执行的，出站 Out 是按照加入 pipeline 的倒序执行的，注意出站处理器 必须向channel通道中写入数据，才会触发出站的handler
示例
public static void main(String[] args) {
new ServerBootstrap()
.group(new NioEventLoopGroup())
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast("h1", new ChannelInboundHandlerAdapter() { // 入站
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.debug("1");
super.channelRead(ctx, msg);
}
});
pipeline.addLast("h2", new ChannelInboundHandlerAdapter() { // 入站
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.debug("2");
super.channelRead(ctx, msg);
// 向 channel 通道中写入数据
// ctx.alloc().buffer() 分配一个 byteBuf
ch.writeAndFlush(ctx.alloc().buffer().writeBytes("hello".getBytes(StandardCharsets.UTF_8)));
}
});
// 出站时，必须向 channel 通道中写入数据，才会触发出站
pipeline.addLast("h3", new ChannelOutboundHandlerAdapter() { // 出站
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.debug("3");
super.write(ctx, msg, promise);
}
});
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);
}
});
}
})
.bind(8080);
}打印顺序，1 2 4 3
———
上面说了 必须向channel通道中写入数据，才会触发出站的handler，注意点 触发出站的方向有两种
- ch.writeAndFlush 向channel通道中写入数据，pipeline会向下查找Out出站的handler
- ctx.writeAndFlush 向channel通道中写入数据，pipeline会向上查找Out出站的handler
介绍一个pipeline调试工具 EmbeddedChannel
public static void main(String[] args) {
ChannelInboundHandlerAdapter h1 = new ChannelInboundHandlerAdapter() {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.debug("1");
super.channelRead(ctx, msg);
}
};
ChannelInboundHandlerAdapter h2 = new ChannelInboundHandlerAdapter() {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.debug("2");
super.channelRead(ctx, msg);
}
};
ChannelOutboundHandlerAdapter h3 = new ChannelOutboundHandlerAdapter() {
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.debug("3");
super.write(ctx, msg, promise);
}
};
ChannelOutboundHandlerAdapter h4 = new ChannelOutboundHandlerAdapter() {
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.debug("4");
super.write(ctx, msg, promise);
}
};
EmbeddedChannel channel = new EmbeddedChannel(h1, h2, h3, h4);
// 模拟入站操作
channel.writeInbound(ByteBufAllocator.DEFAULT.buffer().writeBytes("hello".getBytes(StandardCharsets.UTF_8)));
// 模拟出站操作
channel.writeOutbound(ByteBufAllocator.DEFAULT.buffer().writeBytes("hello".getBytes(StandardCharsets.UTF_8)));
}ByteBuf
ByteBuf 是对 nio 中 ByteBuffer 的增强
-
创建
// 创建一个 buf，默认容量 256 字节，支持动态扩容，默认创建直接内存
ByteBuf buf = ByteBufAllocator.DEFAULT.buffer();
// 在 handler 中请使用 ChannelHandlerContext.alloc().buffer(); 创建ByteBuf-
写入
buf.writeBytes("hello".getBytes(StandardCharsets.UTF_8));| 方法签名 | 含义 | 备注 |
|---|---|---|
| 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) | 写入字符串 |
还有一类方法是 set 开头的一系列方法，也可以写入数据，但不会改变写指针位置
-
ByteBuf的扩容规则
如何写入后数据大小未超过 512，则选择下一个 16 的整数倍，例如写入后大小为 12 ，则扩容后 capacity 是 16
如果写入后数据大小超过 512，则选择下一个 2^n，例如写入后大小为 513，则扩容后 capacity 是 2^10=1024（2^9=512 已经不够了）
扩容不能超过 max capacity 会报错，默认的最大容量是整数的最大值
-
读取
// 一次读一个字节
buf.readByte();
log(buf);
// 一次读指定的字节数
buf.readBytes(1);
log(buf);
// 标记一个位置
buf.markReaderIndex();
buf.readByte();
log(buf);
// 把读指针重置到标记的位置重新读
buf.resetReaderIndex();
buf.readByte();
log(buf);
// 还有一系列读基本类型的
buf.writeInt(1);
// 还有一系列 get 的读方法，不会改变读指针的位置
buf.getByte(1);-
调试 ByteBuf 的工具
private static void log(ByteBuf buffer) {
int length = buffer.readableBytes();
int rows = length / 16 + (length % 15 == 0 ? 0 : 1) + 4;
StringBuilder buf = new StringBuilder(rows * 80 * 2)
.append("read index:").append(buffer.readerIndex())
.append(" write index:").append(buffer.writerIndex())
.append(" capacity:").append(buffer.capacity())
.append(NEWLINE);
appendPrettyHexDump(buf, buffer);
System.out.println(buf.toString());
}-
直接内存、堆内存
直接内存创建和销毁的代价昂贵，但读写性能高（少一次内存复制），适合配合池化功能一起用
直接内存对 GC 压力小，因为这部分内存不受 JVM 垃圾回收的管理，但也要注意及时主动释放
// 创建池化基于直接内存的 ByteBuf，Pooled表示池化
ByteBuf directBuffer = ByteBufAllocator.DEFAULT.directBuffer(10);
System.err.println(directBuffer.getClass());
// 输出：class io.netty.buffer.PooledUnsafeDirectByteBuf
// 创建池化基于堆的 ByteBuf
ByteBuf heapBuffer = ByteBufAllocator.DEFAULT.heapBuffer(10);
System.err.println(heapBuffer.getClass());
// 输出：class io.netty.buffer.PooledUnsafeHeapByteBuf通过环境变量，来指定 DEFAULT 时是使用直接内存或者堆内存，true 堆内存，false 直接内存，默认 false
-Dio.netty.noPreferDirect=true
-
池化 vs 非池化
池化的最大意义在于可以重用 ByteBuf，优点有
没有池化，则每次都得创建新的 ByteBuf 实例，这个操作对直接内存代价昂贵，就算是堆内存，也会增加 GC 压力
有了池化，则可以重用池中 ByteBuf 实例，并且采用了与 jemalloc 类似的内存分配算法提升分配效率
高并发时，池化功能更节约内存，减少内存溢出的可能
池化功能是否开启，可以通过下面的系统环境变量来设置，Android平台默认非池化，其他平台默认池化
-Dio.netty.allocator.type={unpooled|pooled}
-
ByteBuf 的组成
ByteBuf 由四部分组成，最开始读写指针都在0位置，已经读过的字节是废弃字节
-
ByteBuf 的内存释放
由于 Netty 中有堆外内存的 ByteBuf 实现，堆外内存最好是手动来释放，而不是等 GC 垃圾回收。
UnpooledHeapByteBuf (非池化堆内存) 使用的是 JVM 内存，只需等 GC 回收内存即可
UnpooledDirectByteBuf (非池化直接内存) 使用的就是直接内存了，需要特殊的方法来回收内存
PooledByteBuf 和它的子类使用了池化机制，需要更复杂的规则来回收内存
retain & release
Netty 这里采用了引用计数法来控制回收内存（什么是引用计数法查看JVM-垃圾回收-如何判断对象为垃圾对象），每个 ByteBuf 都实现了 ReferenceCounted 接口
每个 ByteBuf 对象的初始计数为 1
调用 release 方法计数减 1，如果计数为 0，ByteBuf 内存被回收
调用 retain 方法计数加 1，表示调用者没用完之前，其它 handler 即使调用了 release 也不会造成回收
当计数为 0 时，底层内存会被回收，这时即使 ByteBuf 对象还在，其各个方法均无法正常使用
在pipeline中有两个netty提供的两个默认的handler，head(头)、tail(尾)，当ByteBuf传到这两个handler中时候，它们会做ByteBuf的释放工作。注意：如果ByteBuf没有传到头尾中 (传到头尾中的是处理过ByteBuf后的Object对象)，那么需要我们手动调用 buf.release() 方法。
零拷贝
-
slice
- slice 零拷贝的体现之一，对原始 ByteBuf 进行切片成多个 ByteBuf，切片后的 ByteBuf 并没有发生内存复制，还是使用原始 ByteBuf 的内存，切片后的 ByteBuf 维护独立的 read，write 指针
- 切片后得到的ByteBuf的最大容量就是切片的大小，因此切片得到的ByteBuf不能write
- 如果原有的ByteBuf进行了release释放内存操作，切片也会收到影响，可以使用retain让引用计数+1
示例：
public static void main(String[] args) {
ByteBuf buf = ByteBufAllocator.DEFAULT.buffer(10);
buf.writeBytes(new byte[] {'a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j'});
// buf的数据是 abcdefghij
// 切片
ByteBuf buf1 = buf.slice(0, 5);
buf1.retain();
// buf1 的数据是 abcde
ByteBuf buf2 = buf.slice(5, 5);
// buf2 的数据是 fghij
buf.release();
// 修改 切片的ByteBuf，原来的 ByteBuf 也会发生改变
buf1.setByte(0, 'z');
// buf 的数据也被改变了，变成了 zbcdefghij
// buf1的数据是 zbcde
buf1.release();
}-
duplicate
零拷贝的体现之一，就好比截取了原始 ByteBuf 所有内容，并且没有 max capacity 的限制，也是与原始 ByteBuf 使用同一块底层内存，只是读写指针是独立的
-
copy
会将底层内存数据进行深拷贝，因此无论读写，都与原始 ByteBuf 无关，相当于新创建了一个新内存的ByteBuf对象
-
CompositeByteBuf
将多个ByteBuf合并组合成一个ByteBuf
public static void main(String[] args) {
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});
CompositeByteBuf buffs = ByteBufAllocator.DEFAULT.compositeBuffer();
buffs.addComponents(true, buf1, buf2);
log(buffs);
}-
Unpooled
Unpooled 是一个工具类，类如其名，提供了非池化的 ByteBuf 创建、组合、复制等操作。
-
ByteBuf 的优点
池化 – 可以重用池中 ByteBuf 实例，更节约内存，减少内存溢出的可能
读写指针分离，不需要像 ByteBuffer 一样切换读写模式
可以自动扩容
支持链式调用，使用更流畅
很多地方体现零拷贝，例如 slice、duplicate、CompositeByteBuf
下一章知识点请阅读：Netty进阶
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 举报，一经查实，本站将立刻删除。
文章由极客之音整理，本文链接：https://www.bmabk.com/index.php/post/72530.html
