Netty 入门
了解netty前需要对nio有一定认识,该笔记基础来自bilinbili黑马,在此基础上自己学习的笔记,添加了一些自己的理解
了解java 非阻塞io编程
1. 概述
1.1 Netty 是什么？
Netty is an asynchronous event-driven network application framework
for rapid development of maintainable high performance protocol servers & clients.
Netty 是一个异步(基于多线程)的、基于事件驱动(多路复用的那写事件驱动)的网络应用框架，用于快速开发可维护、高性能的网络服务器和客户端
末尾练习 简记 c/s各组件结构
/**
* 可以简单记忆是这是一个二级的关系
* 通道是双向连接通道,客户端和服务器端都可以进行处理
* 服务器端使用的是ServerSocketChannel 而客户端是SocketChannel
* eventloopgroup 是相当于封装了nio中的各个触发事件
* 而bootstrap启动器中的group方法类似于在nio多路复用上实现boss,worker的细分
* 服务器端接收到了客户端的连接请求后(对应nio编程中的accept事件) childHdander进行处理数据
* 而childhander中处理的是创建连接后的双向NioSocketChannel
* (对等关系)服务端 : 客户端
* ServerSocketChannel -> SocketChanel :SocketChannel
* childhandler : handler
* 俩段各自可以对通道传输的数据进行处理
*/
1.2 Netty 的作者
他还是另一个著名网络应用框架 Mina 的重要贡献者
1.3 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.4 Netty 的优势
- Netty vs NIO，工作量大，bug 多
- 需要自己构建协议
- 解决 TCP 传输问题，如粘包、半包
- epoll 空轮询导致 CPU 100%
- 对 API 进行增强，使之更易用，如 FastThreadLocal => ThreadLocal，ByteBuf => ByteBuffer
- Netty vs 其它网络应用框架
- Mina 由 apache 维护，将来 3.x 版本可能会有较大重构，破坏 API 向下兼容性，Netty 的开发迭代更迅速，API 更简洁、文档更优秀
- 久经考验，16年，Netty 版本
- 2.x 2004
- 3.x 2008
- 4.x 2013
- 5.x 已废弃（没有明显的性能提升，维护成本高）
2. Hello World
2.1 目标
开发一个简单的服务器端和客户端
- 客户端向服务器端发送 hello, world
- 服务器仅接收，不返回
加入依赖
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.39.Final</version>
</dependency>
2.2 服务器端
public class Server {
public static void main(String[] args) {
// 服务启动器
new ServerBootstrap()
.group(new NioEventLoopGroup()) // 1 线程工厂创建多路复用 事件处理组 包含了线程选择器 和事件的具体实现(nio 和nio 也就是bio )
.channel(NioServerSocketChannel.class) // 2选择服务器通道实现
/**
* 处理分工
* slector决定了worker child字模块
* boss 负责接收客户端的连接
* child 负责处理客户端的读写
*/
.childHandler(
new ChannelInitializer<NioSocketChannel>() { // 3 服务器同意连接后的和客户端相连接的通道
protected void initChannel(NioSocketChannel ch) {
//向通道添加具体业务处理
ch.pipeline().addLast(new StringDecoder()); // 5 byteBUf转化字符串
// 6自定义处理类
ch.pipeline().addLast(new SimpleChannelInboundHandler<String>() {
// 通道读事件
@Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) {
//message 就是Stringdecoder转换的结果
System.out.println(msg);
}
});
}
})
.bind(8080);//服务器绑定端口
}
}
代码解读
-
1 处，创建 NioEventLoopGroup，可以简单理解为
线程池 + Selector后面会详细展开 -
2 处，选择服务 Scoket 实现类，其中 NioServerSocketChannel 表示基于 NIO 的服务器端实现，其它实现还有
-
3 处，为啥方法叫 childHandler，是接下来添加的处理器都是给 SocketChannel 用的，而不是给 ServerSocketChannel。ChannelInitializer 处理器（仅执行一次），它的作用是待客户端 SocketChannel 建立连接后，执行 initChannel 以便添加更多的处理器
-
4 处，ServerSocketChannel 绑定的监听端口
-
5 处，SocketChannel 的处理器，解码 ByteBuf => String
-
6 处，SocketChannel 的业务处理器，使用上一个处理器的处理结果
2.3 客户端
public class Client {
public static void main(String[] args) throws InterruptedException {
// 新建客户启动类绑定连接信息
new Bootstrap()
//添加事件eventloop组
.group(new NioEventLoopGroup())
//绑定客户端通道
.channel(NioSocketChannel.class)
// 添加事件处理器
.handler(new ChannelInitializer<NioSocketChannel>() {//通道是启动器中绑定的通道
//初始化方法 建立连接后调用
@Override
protected void initChannel(NioSocketChannel ch) throws Exception {
//服务器那边拿到字节 解码字符串 所以这边需要添加字符处理器 编码
ch.pipeline().addLast(new StringEncoder());//ChannelPipeline 是一个处理输入和输出数据的通道处理链添加处理器
}
})
//客户端配置完成后连接服务器
.connect(new InetSocketAddress("127.0.0.1", 8080))
.sync()//同步方式
.channel()
.writeAndFlush("hello netty hello network")
;
}
}
//核心逻辑
new Bootstrap()
.group(new NioEventLoopGroup()) // 1
.channel(NioSocketChannel.class) // 2
.handler(new ChannelInitializer<Channel>() { // 3
@Override
protected void initChannel(Channel ch) {
ch.pipeline().addLast(new StringEncoder()); // 8
}
})
.connect("127.0.0.1", 8080) // 4
.sync() // 5
.channel() // 6
.writeAndFlush(new Date() + ": hello world!"); // 7
代码解读
-
1 处，创建 NioEventLoopGroup，同 Server
-
2 处，选择客户 Socket 实现类，NioSocketChannel 表示基于 NIO 的客户端实现，其它实现还有
-
3 处，添加 SocketChannel 的处理器，ChannelInitializer 处理器（仅执行一次），它的作用是待客户端 SocketChannel 建立连接后，执行 initChannel 以便添加更多的处理器
-
4 处，指定要连接的服务器和端口
-
5 处，Netty 中很多方法都是异步的，如 connect，这时需要使用 sync 方法等待 connect 建立连接完毕
-
6 处，获取 channel 对象，它即为通道抽象，可以进行数据读写操作
-
7 处，写入消息并清空缓冲区
-
8 处，消息会经过通道 handler 处理，这里是将 String => ByteBuf 发出
-
数据经过网络传输，到达服务器端，服务器端 5 和 6 处的 handler 先后被触发，走完一个流程
2.4 流程梳理
💡 提示
一开始需要树立正确的观念
- 把 channel 理解为数据的通道
- 把 msg 理解为流动的数据，最开始输入是 ByteBuf，但经过 pipeline 的加工，会变成其它类型对象，最后输出又变成 ByteBuf
- 把 handler 理解为数据的处理工序
- 工序有多道，合在一起就是 pipeline，pipeline 负责发布事件（读、读取完成…）传播给每个 handler， handler 对自己感兴趣的事件进行处理（重写了相应事件处理方法）
- handler 分 Inbound 和 Outbound 两类
- 把 eventLoop 理解为处理数据的工人
- 工人可以管理多个 channel 的 io 操作，并且一旦工人负责了某个 channel，就要负责到底（绑定）触发事件的线程安全
- 工人既可以执行 io 操作，也可以进行任务处理，每位工人有任务队列，队列里可以堆放多个 channel 的待处理任务，任务分为普通任务、定时任务
- 工人按照 pipeline 顺序，依次按照 handler 的规划（代码）处理数据，可以为每道工序指定不同的工人
3. 组件
3.1 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
本质就是封装了nio中selector事件触发,new NioEventLoopGroup() 创建了一个基于 NIO 的事件循环组（EventLoopGroup）。Netty 使用事件循环组来处理事件，其中包括事件的选择、处理和调度。
具体来说，NioEventLoopGroup 封装了一组 NIO 的事件循环（EventLoop），每个事件循环负责处理一部分连接的事件。事件循环是 Netty 处理 IO 事件的核心机制。
在 NIO 中，事件循环主要包括以下几个部分：1.Selector（选择器）： 负责监听通道上的事件，例如连接就绪、读就绪、写就绪等。
2.事件处理器： 在事件发生时，调用注册的事件处理器来处理具体的业务逻辑。这些事件处理器通常是由用户自定义的。
3.调度器： 控制事件的调度，确保事件按照正确的顺序被处理。通过将这些组件封装在 NioEventLoopGroup 中，Netty 隐藏了底层事件循环的复杂性，提供了一种高度抽象的方式来处理异步事件。每个 NioEventLoop（事件循环）都在自己的线程中运行，负责管理一组通道，并处理这些通道上的事件。
正是这样。NioEventLoopGroup 通过底层的 NIO 机制来监听并处理多种事件，包括：1.Accept 事件： 当有新的连接请求时，Selector 会触发 Accept 事件，表示可以接受新的客户端连接。
2.Connect 事件： 当客户端发起连接请求，Selector 会触发 Connect 事件，表示连接已经建立。
3.Read 事件： 当通道中有数据可读时，Selector 会触发 Read 事件，表示可以读取数据。
4.Write 事件： 当通道可写时，Selector 会触发 Write 事件，表示可以向通道写入数据。NioEventLoopGroup 中的每个 NioEventLoop 负责处理一部分连接的事件。这些事件会被委托给注册在相应通道上的事件处理器进行处理。例如，你在代码中添加的 ChannelInitializer 中的 initChannel 方法中的处理器，就会处理连接建立后的读写事件。
这整个过程构成了事件驱动的异步编程模型，使得 Netty 能够高效地处理大量并发连接，而不需要为每个连接分配一个线程。这对于构建高性能、可扩展的网络应用至关重要。
以一个简单的实现为例：
// 内部创建了两个 EventLoop, 每个 EventLoop 维护一个线程
DefaultEventLoopGroup group = new DefaultEventLoopGroup(2);
System.out.println(group.next());
System.out.println(group.next());
System.out.println(group.next());
默认空参线程
输出
io.netty.channel.DefaultEventLoop@60f82f98
io.netty.channel.DefaultEventLoop@35f983a6
io.netty.channel.DefaultEventLoop@60f82f98
也可以使用 for 循环
DefaultEventLoopGroup group = new DefaultEventLoopGroup(2);
for (EventExecutor eventLoop : group) {
System.out.println(eventLoop);
}
输出
io.netty.channel.DefaultEventLoop@60f82f98
io.netty.channel.DefaultEventLoop@35f983a6
基本api
@Slf4j
public class TestEventLoop {
public static void main(String[] args) {
NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup(2);//io事件循环组
DefaultEventLoop defaultEventLoop = new DefaultEventLoop();//默认事件循环组 ,定时任务 普通网络事件任务
System.out.println("当前线程"+Thread.currentThread().getName());
// 获取下一个事件循环对象 (也就是另一个线程)
// d
//给事件循环组添加任务
// 不在同一个线程 肯定是异步执行
eventLoopGroup.next().submit(()->{
log.debug("ok");
});
eventLoopGroup.next().submit(()->{
log.debug("ok");
});
//也是提交任务的api 和submit一样
eventLoopGroup.execute(()->{
log.debug("ok");
});
//执行定时任务
eventLoopGroup.scheduleAtFixedRate(()->{
log.debug("当前定时任务是每一秒打印一次");
},0,1,java.util.concurrent.TimeUnit.SECONDS);
}
}
💡 优雅关闭
优雅关闭 shutdownGracefully 方法。该方法会首先切换 EventLoopGroup 到关闭状态从而拒绝新的任务的加入，然后在任务队列的任务都处理完成后，停止线程的运行。从而确保整体应用是在正常有序的状态下退出的
演示 NioEventLoop 处理 io 事件
服务器端两个 nio worker 工人
new ServerBootstrap()
.group(new NioEventLoopGroup(1), new NioEventLoopGroup(2))
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) {
ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
ByteBuf byteBuf = msg instanceof ByteBuf ? ((ByteBuf) msg) : null;
if (byteBuf != null) {
byte[] buf = new byte[16];
ByteBuf len = byteBuf.readBytes(buf, 0, byteBuf.readableBytes());
log.debug(new String(buf));
}
}
});
}
}).bind(8080).sync();
客户端，启动三次，分别修改发送字符串为 zhangsan（第一次），lisi（第二次），wangwu（第三次）
public static void main(String[] args) throws InterruptedException {
Channel channel = new Bootstrap()
.group(new NioEventLoopGroup(1))
.handler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) throws Exception {
System.out.println("init...");
ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
}
})
.channel(NioSocketChannel.class).connect("localhost", 8080)
.sync()
.channel();
channel.writeAndFlush(ByteBufAllocator.DEFAULT.buffer().writeBytes("wangwu".getBytes()));
Thread.sleep(2000);
channel.writeAndFlush(ByteBufAllocator.DEFAULT.buffer().writeBytes("wangwu".getBytes()));
最后输出
22:03:34 [DEBUG] [nioEventLoopGroup-3-1] c.i.o.EventLoopTest - zhangsan
22:03:36 [DEBUG] [nioEventLoopGroup-3-1] c.i.o.EventLoopTest - zhangsan
22:05:36 [DEBUG] [nioEventLoopGroup-3-2] c.i.o.EventLoopTest - lisi
22:05:38 [DEBUG] [nioEventLoopGroup-3-2] c.i.o.EventLoopTest - lisi
22:06:09 [DEBUG] [nioEventLoopGroup-3-1] c.i.o.EventLoopTest - wangwu
22:06:11 [DEBUG] [nioEventLoopGroup-3-1] c.i.o.EventLoopTest - wangwu
可以看到两个工人轮流处理 channel，但工人与 channel 之间进行了绑定
再增加两个非 nio 工人
DefaultEventLoopGroup normalWorkers = new DefaultEventLoopGroup(2);
new ServerBootstrap()
.group(new NioEventLoopGroup(1), new NioEventLoopGroup(2))
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) {
ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
ch.pipeline().addLast(normalWorkers,"myhandler",
new ChannelInboundHandlerAdapter() {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
ByteBuf byteBuf = msg instanceof ByteBuf ? ((ByteBuf) msg) : null;
if (byteBuf != null) {
byte[] buf = new byte[16];
ByteBuf len = byteBuf.readBytes(buf, 0, byteBuf.readableBytes());
log.debug(new String(buf));
}
}
});
}
}).bind(8080).sync();
客户端代码不变，启动三次，分别修改发送字符串为 zhangsan（第一次），lisi（第二次），wangwu（第三次）
输出
22:19:48 [DEBUG] [nioEventLoopGroup-4-1] i.n.h.l.LoggingHandler - [id: 0x251562d5, L:/127.0.0.1:8080 - R:/127.0.0.1:52588] REGISTERED
22:19:48 [DEBUG] [nioEventLoopGroup-4-1] i.n.h.l.LoggingHandler - [id: 0x251562d5, L:/127.0.0.1:8080 - R:/127.0.0.1:52588] ACTIVE
22:19:48 [DEBUG] [nioEventLoopGroup-4-1] i.n.h.l.LoggingHandler - [id: 0x251562d5, L:/127.0.0.1:8080 - R:/127.0.0.1:52588] READ: 8B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 7a 68 61 6e 67 73 61 6e |zhangsan |
+--------+-------------------------------------------------+----------------+
22:19:48 [DEBUG] [nioEventLoopGroup-4-1] i.n.h.l.LoggingHandler - [id: 0x251562d5, L:/127.0.0.1:8080 - R:/127.0.0.1:52588] READ COMPLETE
22:19:48 [DEBUG] [defaultEventLoopGroup-2-1] c.i.o.EventLoopTest - zhangsan
22:19:50 [DEBUG] [nioEventLoopGroup-4-1] i.n.h.l.LoggingHandler - [id: 0x251562d5, L:/127.0.0.1:8080 - R:/127.0.0.1:52588] READ: 8B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 7a 68 61 6e 67 73 61 6e |zhangsan |
+--------+-------------------------------------------------+----------------+
22:19:50 [DEBUG] [nioEventLoopGroup-4-1] i.n.h.l.LoggingHandler - [id: 0x251562d5, L:/127.0.0.1:8080 - R:/127.0.0.1:52588] READ COMPLETE
22:19:50 [DEBUG] [defaultEventLoopGroup-2-1] c.i.o.EventLoopTest - zhangsan
22:20:24 [DEBUG] [nioEventLoopGroup-4-2] i.n.h.l.LoggingHandler - [id: 0x94b2a840, L:/127.0.0.1:8080 - R:/127.0.0.1:52612] REGISTERED
22:20:24 [DEBUG] [nioEventLoopGroup-4-2] i.n.h.l.LoggingHandler - [id: 0x94b2a840, L:/127.0.0.1:8080 - R:/127.0.0.1:52612] ACTIVE
22:20:25 [DEBUG] [nioEventLoopGroup-4-2] i.n.h.l.LoggingHandler - [id: 0x94b2a840, L:/127.0.0.1:8080 - R:/127.0.0.1:52612] READ: 4B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 6c 69 73 69 |lisi |
+--------+-------------------------------------------------+----------------+
22:20:25 [DEBUG] [nioEventLoopGroup-4-2] i.n.h.l.LoggingHandler - [id: 0x94b2a840, L:/127.0.0.1:8080 - R:/127.0.0.1:52612] READ COMPLETE
22:20:25 [DEBUG] [defaultEventLoopGroup-2-2] c.i.o.EventLoopTest - lisi
22:20:27 [DEBUG] [nioEventLoopGroup-4-2] i.n.h.l.LoggingHandler - [id: 0x94b2a840, L:/127.0.0.1:8080 - R:/127.0.0.1:52612] READ: 4B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 6c 69 73 69 |lisi |
+--------+-------------------------------------------------+----------------+
22:20:27 [DEBUG] [nioEventLoopGroup-4-2] i.n.h.l.LoggingHandler - [id: 0x94b2a840, L:/127.0.0.1:8080 - R:/127.0.0.1:52612] READ COMPLETE
22:20:27 [DEBUG] [defaultEventLoopGroup-2-2] c.i.o.EventLoopTest - lisi
22:20:38 [DEBUG] [nioEventLoopGroup-4-1] i.n.h.l.LoggingHandler - [id: 0x79a26af9, L:/127.0.0.1:8080 - R:/127.0.0.1:52625] REGISTERED
22:20:38 [DEBUG] [nioEventLoopGroup-4-1] i.n.h.l.LoggingHandler - [id: 0x79a26af9, L:/127.0.0.1:8080 - R:/127.0.0.1:52625] ACTIVE
22:20:38 [DEBUG] [nioEventLoopGroup-4-1] i.n.h.l.LoggingHandler - [id: 0x79a26af9, L:/127.0.0.1:8080 - R:/127.0.0.1:52625] READ: 6B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 77 61 6e 67 77 75 |wangwu |
+--------+-------------------------------------------------+----------------+
22:20:38 [DEBUG] [nioEventLoopGroup-4-1] i.n.h.l.LoggingHandler - [id: 0x79a26af9, L:/127.0.0.1:8080 - R:/127.0.0.1:52625] READ COMPLETE
22:20:38 [DEBUG] [defaultEventLoopGroup-2-1] c.i.o.EventLoopTest - wangwu
22:20:40 [DEBUG] [nioEventLoopGroup-4-1] i.n.h.l.LoggingHandler - [id: 0x79a26af9, L:/127.0.0.1:8080 - R:/127.0.0.1:52625] READ: 6B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 77 61 6e 67 77 75 |wangwu |
+--------+-------------------------------------------------+----------------+
22:20:40 [DEBUG] [nioEventLoopGroup-4-1] i.n.h.l.LoggingHandler - [id: 0x79a26af9, L:/127.0.0.1:8080 - R:/127.0.0.1:52625] READ COMPLETE
22:20:40 [DEBUG] [defaultEventLoopGroup-2-1] c.i.o.EventLoopTest - wangwu
自己进一步细分eventloop
/**
* 处理io事件
*/
@Slf4j
public class IOEventLoop {
public static void main(String[] args) {
DefaultEventLoopGroup executors = new DefaultEventLoopGroup();//用于做work操作做某些任务的循环组
new ServerBootstrap()
//进一步划分职责 boss 和具体执行worker
/**
* todo1
* 所有事件交给一个事件循环组 导致 boss 处理新连接的请求过慢 从而导致 连接超时
* * 因此需要将 boss 和 worker 分开
* @param1 boss 负责监听端口，accept 新连接的
* @param2 worker 负责具体的处理
* todo2 但具体处理的worker处理事件过长
* 所有的事件都由worker来处理 比如 读写事件 连接事件
* 心跳事件 等等 这些事件 可能由不同的worker来处理
*/
.group(new NioEventLoopGroup(),new NioEventLoopGroup())
.channel(NioServerSocketChannel.class)
.childHandler(
// 通道初始化器
new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) throws Exception {
// 处理链路添加具体处理器处理 参数只有处理器的时候 由worker线程执行
ch.pipeline().addLast("hander1",new ChannelInboundHandlerAdapter(){//inbound 读取事件 out 输出写事件
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
// 由于没有添加字符处理器 接收的数据是bytebuf bytebuffer的增强
System.out.println("服务器触发读取事件");
ByteBuf byteBuf = (ByteBuf) msg;
log.info("接收的数据转字符{}",byteBuf.toString(CharsetUtil.UTF_8));
//处理完毕后 利用上下文对象 将消息给下一个处理器二次加工
ctx.fireChannelRead(msg);
}
});
/**
* @parm1 指定哪个事件循环的线程触发
* @parm2 处理起名
* @parm3 处理器
*/
ch.pipeline().addLast(executors,"hander2",new ChannelInboundHandlerAdapter(){//inbound 读取事件 out 输出写事件
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
// 由于没有添加字符处理器 接收的数据是bytebuf bytebuffer的增强
System.out.println("handler2触发读取事件");
ByteBuf byteBuf = (ByteBuf) msg;
log.info("接收的数据转字符{}",byteBuf.toString(CharsetUtil.UTF_8));
}
});//
}
})
.bind(8044);
}
}
可以看到，nio 工人和 非 nio 工人也分别绑定了 channel（LoggingHandler 由 nio 工人执行，而我们自己的 handler 由非 nio 工人执行）
💡 handler 执行中如何换人？
关键代码 io.netty.channel.AbstractChannelHandlerContext#invokeChannelRead()
static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {
final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next);
// 下一个 handler 的事件循环是否与当前的事件循环是同一个线程
EventExecutor executor = next.executor();
// 是，直接调用
if (executor.inEventLoop()) {
next.invokeChannelRead(m);
}
// 不是，将要执行的代码作为任务提交给下一个事件循环处理（换人）
else {
//通过封装在任务对象传递给另一个线程使用
executor.execute(new Runnable() {
@Override
public void run() {
next.invokeChannelRead(m);
}
});
}
}
- 如果两个 handler 绑定的是同一个线程，那么就直接调用
- 否则，把要调用的代码封装为一个任务对象，由下一个 handler 的线程来调用
演示 NioEventLoop 处理普通任务
NioEventLoop 除了可以处理 io 事件，同样可以向它提交普通任务
NioEventLoopGroup nioWorkers = new NioEventLoopGroup(2);
log.debug("server start...");
Thread.sleep(2000);
nioWorkers.execute(()->{
log.debug("normal task...");
});
输出
22:30:36 [DEBUG] [main] c.i.o.EventLoopTest2 - server start...
22:30:38 [DEBUG] [nioEventLoopGroup-2-1] c.i.o.EventLoopTest2 - normal task...
可以用来执行耗时较长的任务
演示 NioEventLoop 处理定时任务
NioEventLoopGroup nioWorkers = new NioEventLoopGroup(2);
log.debug("server start...");
Thread.sleep(2000);
nioWorkers.scheduleAtFixedRate(() -> {
log.debug("running...");
}, 0, 1, TimeUnit.SECONDS);
输出
22:35:15 [DEBUG] [main] c.i.o.EventLoopTest2 - server start...
22:35:17 [DEBUG] [nioEventLoopGroup-2-1] c.i.o.EventLoopTest2 - running...
22:35:18 [DEBUG] [nioEventLoopGroup-2-1] c.i.o.EventLoopTest2 - running...
22:35:19 [DEBUG] [nioEventLoopGroup-2-1] c.i.o.EventLoopTest2 - running...
22:35:20 [DEBUG] [nioEventLoopGroup-2-1] c.i.o.EventLoopTest2 - running...
...
可以用来执行定时任务
3.2 Channel
channel 的主要作用
- close() 可以用来关闭 channel
- closeFuture() 用来处理 channel 的关闭
- sync 方法作用是同步等待 channel 关闭
- 而 addListener 方法是异步等待 channel 关闭
- pipeline() 方法添加处理器
- write() 方法将数据写入
- writeAndFlush() 方法将数据写入并刷出
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)
.sync()
.channel()
.writeAndFlush(new Date() + ": hello world!");
现在把它拆开来看
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); // 1
channelFuture.sync().channel().writeAndFlush(new Date() + ": hello world!");
- 1 处返回的是 ChannelFuture 对象，它的作用是利用 channel() 方法来获取 Channel 对象
- //同步方式 netty的事件组默认多个线程 如果不采用同步连接会出现数据不同步问题
/**
* 这个ChannelFuture是 Netty 框架中的一个组件，用于表示异步操作的执行结果。
* proimise ,future,callback都是表示异步操作
* 在这个例子中，ChannelFuture用于表示客户端与服务器之间的连接操作。
*
*ChannelFuture的sync()方法用于同步等待连接操作的完成。当调用sync()时，它会阻塞主线程，直到连接操作完成或者抛出异常。
* 在这个例子中，sync()的调用确保了主线程在连接成功之前不会继续执行，从而避免了数据发送失败的问题。
*/
注意 connect 方法是异步的，意味着不等连接建立，方法执行就返回了。因此 channelFuture 对象中不能【立刻】获得到正确的 Channel 对象
实验如下：
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.sync(); // 2该方法异步非阻塞 多线程 nio 执行到这个代码是主线程发起 然后执行连接任务的是循环组的线程
//如果不适用同步方法来阻塞线程 那么事件循环组还没连接,主线程却直接获取通道 就会导致数据发送失败
System.out.println(channelFuture.channel()); // 3
- 执行到 1 时，连接未建立，打印
[id: 0x2e1884dd] - 执行到 2 时，sync 方法是同步等待连接建立完成
- 执行到 3 时，连接肯定建立了，打印
[id: 0x2e1884dd, L:/127.0.0.1:57191 - R:/127.0.0.1:8080]
除了用 sync 方法可以让异步操作同步以外，还可以使用回调的方式：
public class Client {
public static void main(String[] args) throws InterruptedException {
// 新建客户启动类绑定连接信息
ChannelFuture future = new Bootstrap()
//添加事件eventloop组
.group(new NioEventLoopGroup())
//绑定客户端通道
.channel(NioSocketChannel.class)
// 添加事件处理器
.handler(new ChannelInitializer<NioSocketChannel>() {//通道是启动器中绑定的通道
//初始化方法 建立连接后调用
@Override
protected void initChannel(NioSocketChannel ch) throws Exception {
//服务器那边拿到字节 解码字符串 所以这边需要添加字符处理器 编码
ch.pipeline().addLast(new StringEncoder());//ChannelPipeline 是一个处理输入和输出数据的通道处理链添加处理器
}
})
//客户端配置完成后连接服务器
.connect(new InetSocketAddress("127.0.0.1", 8080))//该方法异步非阻塞 多线程 nio 执行到这个代码是主线程发起 然后执行连接任务的是循环组的线程
//如果不适用同步方法来阻塞线程 那么事件循环组还没连接,主线程却直接获取通道 就会导致数据发送失败
//同步方式
// .sync()
// future.channel()
// .writeAndFlush("hello netty hello network"); //写入缓冲区
;
//同步方式 netty的事件组默认多个线程 如果不采用同步连接会出现数据不同步问题
/**
* 这个 `ChannelFuture` 是 Netty 框架中的一个组件，用于表示异步操作的执行结果。
* proimise ,future,callback都是表示异步操作
* 在这个例子中，`ChannelFuture` 用于表示客户端与服务器之间的连接操作。
*
* `ChannelFuture` 的 `sync()` 方法用于同步等待连接操作的完成。当调用 `sync()` 时，它会阻塞主线程，直到连接操作完成或者抛出异常。
* 在这个例子中，`sync()` 的调用确保了主线程在连接成功之前不会继续执行，从而避免了数据发送失败的问题。
*/
// 异步方式
future.addListener(new ChannelFutureListener(){
//实现回调函数
@Override
public void operationComplete(ChannelFuture future) throws Exception {
// 当nio中的连接线程建立后 nio的其他线程在使用该方法
Channel channel = future.channel();
channel.writeAndFlush("async hello netty");
}
});
}
}
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((ChannelFutureListener) future -> {
System.out.println(future.channel()); // 2
});
- 执行到 1 时，连接未建立，打印
[id: 0x749124ba] - ChannelFutureListener 会在连接建立时被调用（其中 operationComplete 方法），因此执行到 2 时，连接肯定建立了，打印
[id: 0x749124ba, L:/127.0.0.1:57351 - R:/127.0.0.1:8080]
CloseFuture
@Slf4j
public class InputChannelClient {
public static void main(String[] args) {
ChannelFuture future = new Bootstrap()
.channel(NioSocketChannel.class)
.group(new NioEventLoopGroup())
.handler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) throws Exception {
ch.pipeline().addLast(new StringEncoder());
// 日志处理器
ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
}
})
.connect("localhost", 8080);
// 异步监听
future.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
Channel channel = future.channel();
new Thread(()->{
Scanner scanner = new Scanner(System.in);
while (true) {
String line = scanner.nextLine();
if (line.equals("q")){
// channel关闭后 回调函数不在执行 也就是说如果while循环后有逻辑 不会执行
channel.close();//close 也是异步方法 需要考虑异步处理线程的速度
log.info("客户端退出 channel closed");//不能在这里善后
break;
}
else //不是退出操作 写入通道
{
channel.writeAndFlush(line);
}
}
},"input").start();
}
});
}
}
其中的通道关闭close方法也是异步非阻塞方法会出现通道channel没有关闭但是但是继续向下执行的情况,可以看到·这里关闭的线程和执行任务的线程有错误
需要进行改进使用到了另一个future结果处理
@Slf4j
public class InputChannelClient {
public static void main(String[] args) {
NioEventLoopGroup group = new NioEventLoopGroup();
ChannelFuture future = new Bootstrap()
.channel(NioSocketChannel.class)
.group(group)
.handler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) throws Exception {
ch.pipeline().addLast(new StringEncoder());
// 日志处理器
ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
}
})
.connect("localhost", 8080);
// 异步监听
future.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
Channel channel = future.channel();
new Thread(()->{
Scanner scanner = new Scanner(System.in);
while (true) {
String line = scanner.nextLine();
if (line.equals("q")){
// channel关闭后 回调函数不在执行 也就是说如果while循环后有逻辑 不会执行
channel.close();//close 也是异步方法 需要考虑异步处理线程的速度
break;
}
else //不是退出操作 写入通道
{
channel.writeAndFlush(line);
}
}
},"input").start();
//获取关闭结果处理
ChannelFuture closeFuture = channel.closeFuture();
closeFuture.addListener(new ChannelFutureListener() {
//调用关闭channel close api的线程触发该会带哦
@Override
public void operationComplete(ChannelFuture future) throws Exception {
log.info("客户端通道关闭");
group.shutdownGracefully();//停止事件循环组继续循环遍历nio线程任务
}
});
}
});
}
}
处理结果
💡 异步提升的是什么
-
有些同学看到这里会有疑问：为什么不在一个线程中去执行建立连接、去执行关闭 channel，那样不是也可以吗？非要用这么复杂的异步方式：比如一个线程发起建立连接，另一个线程去真正建立连接
-
还有同学会笼统地回答，因为 netty 异步方式用了多线程、多线程就效率高。其实这些认识都比较片面，多线程和异步所提升的效率并不是所认为的
思考下面的场景，4 个医生给人看病，每个病人花费 20 分钟，而且医生看病的过程中是以病人为单位的，一个病人看完了，才能看下一个病人。假设病人源源不断地来，可以计算一下 4 个医生一天工作 8 小时，处理的病人总数是：4 * 8 * 3 = 96
经研究发现，看病可以细分为四个步骤，经拆分后每个步骤需要 5 分钟，如下
因此可以做如下优化，只有一开始，医生 2、3、4 分别要等待 5、10、15 分钟才能执行工作，但只要后续病人源源不断地来，他们就能够满负荷工作，并且处理病人的能力提高到了 4 * 8 * 12 效率几乎是原来的四倍
要点
- 单线程没法异步提高效率，必须配合多线程、多核 cpu 才能发挥异步的优势
- 异步并没有缩短响应时间，反而有所增加
- 合理进行任务拆分，也是利用异步的关键
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 | - | - | 设置失败结果 |
例1
同步处理任务成功
DefaultEventLoop eventExecutors = new DefaultEventLoop();
DefaultPromise<Integer> promise = new DefaultPromise<>(eventExecutors);
eventExecutors.execute(()->{
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
log.debug("set success, {}",10);
promise.setSuccess(10);
});
log.debug("start...");
log.debug("{}",promise.getNow()); // 还没有结果
log.debug("{}",promise.get());
输出
11:51:53 [DEBUG] [main] c.i.o.DefaultPromiseTest2 - start...
11:51:53 [DEBUG] [main] c.i.o.DefaultPromiseTest2 - null
11:51:54 [DEBUG] [defaultEventLoop-1-1] c.i.o.DefaultPromiseTest2 - set success, 10
11:51:54 [DEBUG] [main] c.i.o.DefaultPromiseTest2 - 10
例2
异步处理任务成功
DefaultEventLoop eventExecutors = new DefaultEventLoop();
DefaultPromise<Integer> promise = new DefaultPromise<>(eventExecutors);
// 设置回调，异步接收结果
promise.addListener(future -> {
// 这里的 future 就是上面的 promise
log.debug("{}",future.getNow());
});
// 等待 1000 后设置成功结果
eventExecutors.execute(()->{
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
log.debug("set success, {}",10);
promise.setSuccess(10);
});
log.debug("start...");
输出
11:49:30 [DEBUG] [main] c.i.o.DefaultPromiseTest2 - start...
11:49:31 [DEBUG] [defaultEventLoop-1-1] c.i.o.DefaultPromiseTest2 - set success, 10
11:49:31 [DEBUG] [defaultEventLoop-1-1] c.i.o.DefaultPromiseTest2 - 10
例3
同步处理任务失败 - sync & get
DefaultEventLoop eventExecutors = new DefaultEventLoop();
DefaultPromise<Integer> promise = new DefaultPromise<>(eventExecutors);
eventExecutors.execute(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
RuntimeException e = new RuntimeException("error...");
log.debug("set failure, {}", e.toString());
promise.setFailure(e);
});
log.debug("start...");
log.debug("{}", promise.getNow());
promise.get(); // sync() 也会出现异常，只是 get 会再用 ExecutionException 包一层异常
输出
12:11:07 [DEBUG] [main] c.i.o.DefaultPromiseTest2 - start...
12:11:07 [DEBUG] [main] c.i.o.DefaultPromiseTest2 - null
12:11:08 [DEBUG] [defaultEventLoop-1-1] c.i.o.DefaultPromiseTest2 - set failure, java.lang.RuntimeException: error...
Exception in thread "main" java.util.concurrent.ExecutionException: java.lang.RuntimeException: error...
at io.netty.util.concurrent.AbstractFuture.get(AbstractFuture.java:41)
at com.itcast.oio.DefaultPromiseTest2.main(DefaultPromiseTest2.java:34)
Caused by: java.lang.RuntimeException: error...
at com.itcast.oio.DefaultPromiseTest2.lambda$main$0(DefaultPromiseTest2.java:27)
at io.netty.channel.DefaultEventLoop.run(DefaultEventLoop.java:54)
at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:918)
at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
at java.lang.Thread.run(Thread.java:745)
例4
同步处理任务失败 - await
DefaultEventLoop eventExecutors = new DefaultEventLoop();
DefaultPromise<Integer> promise = new DefaultPromise<>(eventExecutors);
eventExecutors.execute(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
RuntimeException e = new RuntimeException("error...");
log.debug("set failure, {}", e.toString());
promise.setFailure(e);
});
log.debug("start...");
log.debug("{}", promise.getNow());
promise.await(); // 与 sync 和 get 区别在于，不会抛异常
log.debug("result {}", (promise.isSuccess() ? promise.getNow() : promise.cause()).toString());
输出
12:18:53 [DEBUG] [main] c.i.o.DefaultPromiseTest2 - start...
12:18:53 [DEBUG] [main] c.i.o.DefaultPromiseTest2 - null
12:18:54 [DEBUG] [defaultEventLoop-1-1] c.i.o.DefaultPromiseTest2 - set failure, java.lang.RuntimeException: error...
12:18:54 [DEBUG] [main] c.i.o.DefaultPromiseTest2 - result java.lang.RuntimeException: error...
例5
异步处理任务失败
DefaultEventLoop eventExecutors = new DefaultEventLoop();
DefaultPromise<Integer> promise = new DefaultPromise<>(eventExecutors);
promise.addListener(future -> {
log.debug("result {}", (promise.isSuccess() ? promise.getNow() : promise.cause()).toString());
});
eventExecutors.execute(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
RuntimeException e = new RuntimeException("error...");
log.debug("set failure, {}", e.toString());
promise.setFailure(e);
});
log.debug("start...");
输出
12:04:57 [DEBUG] [main] c.i.o.DefaultPromiseTest2 - start...
12:04:58 [DEBUG] [defaultEventLoop-1-1] c.i.o.DefaultPromiseTest2 - set failure, java.lang.RuntimeException: error...
12:04:58 [DEBUG] [defaultEventLoop-1-1] c.i.o.DefaultPromiseTest2 - result java.lang.RuntimeException: error...
例6
await 死锁检查
DefaultEventLoop eventExecutors = new DefaultEventLoop();
DefaultPromise<Integer> promise = new DefaultPromise<>(eventExecutors);
eventExecutors.submit(()->{
System.out.println("1");
try {
promise.await();
// 注意不能仅捕获 InterruptedException 异常
// 否则 死锁检查抛出的 BlockingOperationException 会继续向上传播
// 而提交的任务会被包装为 PromiseTask，它的 run 方法中会 catch 所有异常然后设置为 Promise 的失败结果而不会抛出
} catch (Exception e) {
e.printStackTrace();
}
System.out.println("2");
});
eventExecutors.submit(()->{
System.out.println("3");
try {
promise.await();
} catch (Exception e) {
e.printStackTrace();
}
System.out.println("4");
});
输出
1
2
3
4
io.netty.util.concurrent.BlockingOperationException: DefaultPromise@47499c2a(incomplete)
at io.netty.util.concurrent.DefaultPromise.checkDeadLock(DefaultPromise.java:384)
at io.netty.util.concurrent.DefaultPromise.await(DefaultPromise.java:212)
at com.itcast.oio.DefaultPromiseTest.lambda$main$0(DefaultPromiseTest.java:27)
at io.netty.util.concurrent.PromiseTask$RunnableAdapter.call(PromiseTask.java:38)
at io.netty.util.concurrent.PromiseTask.run(PromiseTask.java:73)
at io.netty.channel.DefaultEventLoop.run(DefaultEventLoop.java:54)
at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:918)
at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
at java.lang.Thread.run(Thread.java:745)
io.netty.util.concurrent.BlockingOperationException: DefaultPromise@47499c2a(incomplete)
at io.netty.util.concurrent.DefaultPromise.checkDeadLock(DefaultPromise.java:384)
at io.netty.util.concurrent.DefaultPromise.await(DefaultPromise.java:212)
at com.itcast.oio.DefaultPromiseTest.lambda$main$1(DefaultPromiseTest.java:36)
at io.netty.util.concurrent.PromiseTask$RunnableAdapter.call(PromiseTask.java:38)
at io.netty.util.concurrent.PromiseTask.run(PromiseTask.java:73)
at io.netty.channel.DefaultEventLoop.run(DefaultEventLoop.java:54)
at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:918)
at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
at java.lang.Thread.run(Thread.java:745)
JDK中基础的future
@Slf4j
public class TestJDKFuture {
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService pool = Executors.newFixedThreadPool(3);
Future<Integer> future = pool.submit(() -> {
Thread.sleep(1000);
log.debug("线程池进行逻辑计算");
return 50;
});
//主线程 堵塞
// 主线程获取到线程池中的对象
Integer integer = future.get();
log.debug("主线程获取到结果：{}",integer);
}
}
封装后netty的future
@Slf4j
public class TestNettyFuture {
public static void main(String[] args) throws ExecutionException, InterruptedException {
// 创建EventLoopGroup 事件循环对象
NioEventLoopGroup loopGroup = new NioEventLoopGroup();
// 拿到事件循环对象组中的任务监听对象
EventLoop next = loopGroup.next();
// 执行任务
// 无论哪种方式 future位置结果对象都是提交任务后获得的·
Future<Integer> future = next.submit(() -> {
Thread.sleep(1000);
log.debug("事件监听循环组进行逻辑计算");
return 70;
});
log.info("主线程等待事件监听循环组计算结果");
// 等待计算结果 同步方式
log.debug("同步事件监听循环组计算结果：{}", future.get());
future.addListener(future1 ->{
// 异步处理·输出是其他线程得到了结果输出
log.debug("异步事件监听循环组计算结果：{}", future1.get());//或者getnow (get阻塞)
});
}
}
promise
@Slf4j
public class TestPromise {
public static void main(String[] args) throws ExecutionException, InterruptedException {
NioEventLoopGroup loopGroup = new NioEventLoopGroup();
//promise 对象可以手动创建使用 存放结果容器
DefaultPromise<Integer> promise = new DefaultPromise<>(loopGroup.next());
new Thread(()->{
//主线程调用promise的方法进入该线程
//任意线程线程计算后向promise 设置成功值
log.info("开始计算");
/**
* promise对象可以通过手动计算赋值
* 主线程进行阻塞等待promise结果
*/
try{
promise.setSuccess(80);
}
catch (Exception e){
promise.setFailure(new RuntimeException("计算失败"));
}
}).start();
log.info("等待结果");
log.info("结果是{}",promise.get());
}
}
future和promise
在 Netty 中，Promise 和 Future 是两个关联但不同的概念。它们都与异步操作的结果和状态有关，但在功能和用途上存在一些区别。
1.Future（未来）：
2.意义： 代表一个异步操作的结果或者正在进行的操作的状态。
3.作用： 允许在操作完成之后获取结果，或者注册回调以处理操作完成时的事件。
4.操作： 通常有方法来检查操作是否已经完成（isDone）、等待操作完成（get）等。5.Promise（承诺）：
6.意义： 代表一个异步操作的可写部分，即操作的完成状态。
7.作用： 允许设置操作的结果，完成操作，或者处理操作失败的情况。
8.操作： 提供了设置操作结果的方法（setResult）、完成操作的方法（trySuccess）等。区别：
9.责任划分： Future 主要用于表示异步操作的结果和状态，而 Promise 主要用于操作的状态的改变，即它是 Future 的一个补充，提供了对异步操作结果的设置能力。
10.可写性： Future 是只读的，一旦获得了 Future 对象，你只能通过它获取操作的结果。相比之下，Promise 是可写的，你可以使用 Promise 对象来设置操作的结果。
11.组合： 通常，一个异步操作的结果是由 Future 对象表示的，而这个 Future 对象的可写部分就是对应的 Promise 对象。这种设计使得异步操作的结果和状态能够被清晰地划分和处理。在 Netty 中，ChannelFuture 表示一个异步 I/O 操作的结果，而 ChannelPromise 则是 ChannelFuture 的可写部分，用于设置操作的结果。这种设计使得 Netty 中的异步操作更加灵活和可控。
promise更多的是对future的补充,future是装载异步处理结果的容器,promise就是对其进行异步改写
3.4 Handler & Pipeline
ChannelHandler 用来处理 Channel 上的各种事件，分为入站、出站两种。所有 ChannelHandler 被连成一串，就是 Pipeline
- 入站处理器通常是 ChannelInboundHandlerAdapter 的子类，主要用来读取客户端数据，写回结果
- 出站处理器通常是 ChannelOutboundHandlerAdapter 的子类，主要对写回结果进行加工
打个比喻，每个 Channel 是一个产品的加工车间，Pipeline 是车间中的流水线，ChannelHandler 就是流水线上的各道工序，而后面要讲的 ByteBuf 是原材料，经过很多工序的加工：先经过一道道入站工序，再经过一道道出站工序最终变成产品
先搞清楚顺序，服务端
new ServerBootstrap()
.group(new NioEventLoopGroup())
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<NioSocketChannel>() {
protected void initChannel(NioSocketChannel ch) {
ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
System.out.println(1);
ctx.fireChannelRead(msg); // 1
}
});
ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
System.out.println(2);
ctx.fireChannelRead(msg); // 2
}
});
ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
System.out.println(3);
ctx.channel().write(msg); // 3
}
});
ch.pipeline().addLast(new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg,
ChannelPromise promise) {
System.out.println(4);
ctx.write(msg, promise); // 4
}
});
ch.pipeline().addLast(new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg,
ChannelPromise promise) {
System.out.println(5);
ctx.write(msg, promise); // 5
}
});
ch.pipeline().addLast(new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg,
ChannelPromise promise) {
System.out.println(6);
ctx.write(msg, promise); // 6
}
});
}
})
.bind(8080);
客户端
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)
.addListener((ChannelFutureListener) future -> {
future.channel().writeAndFlush("hello,world");
});
服务器端打印：
1
2
3
6
5
4
可以看到，ChannelInboundHandlerAdapter 是按照 addLast 的顺序执行的，而 ChannelOutboundHandlerAdapter 是按照 addLast 的逆序执行的。ChannelPipeline 的实现是一个 ChannelHandlerContext（包装了 ChannelHandler） 组成的双向链表
- 入站处理器中，ctx.fireChannelRead(msg) 是 （从头向后）调用下一个入站处理器
- 如果注释掉 1 处代码，则仅会打印 1
- 如果注释掉 2 处代码，则仅会打印 1 2
- 3 处的 ctx.channel().write(msg) 会 从尾部开始触发 后续出站处理器的执行
- 如果注释掉 3 处代码，则仅会打印 1 2 3
- 类似的，出站处理器中，ctx.write(msg, promise) 的调用也会 (从尾往前)触发上一个出站处理器
- 如果注释掉 6 处代码，则仅会打印 1 2 3 6
- ctx.channel().write(msg) vs ctx.write(msg)
- 都是触发出站处理器的执行
- ctx.channel().write(msg) 从尾部开始查找出站处理器
- ctx.write(msg) 是从当前节点找上一个出站处理器
- 3 处的 ctx.channel().write(msg) 如果改为 ctx.write(msg) 仅会打印 1 2 3，因为节点3 之前没有其它出站处理器了
- 6 处的 ctx.write(msg, promise) 如果改为 ctx.channel().write(msg) 会打印 1 2 3 6 6 6… 因为 ctx.channel().write() 是从尾部开始查找，结果又是节点6 自己
图1 - 服务端 pipeline 触发的原始流程，图中数字代表了处理步骤的先后次序
3.5 ByteBuf
是对字节数据的封装
1）创建
ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(10);
log(buffer);
上面代码创建了一个默认的 ByteBuf（池化基于直接内存的 ByteBuf），初始容量是 10
输出
read index:0 write index:0 capacity:10
其中 log 方法参考如下
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());
}
2）直接内存 vs 堆内存
可以使用下面的代码来创建池化基于堆的 ByteBuf
ByteBuf buffer = ByteBufAllocator.DEFAULT.heapBuffer(10);
也可以使用下面的代码来创建池化基于直接内存的 ByteBuf
ByteBuf buffer = ByteBufAllocator.DEFAULT.directBuffer(10);
- 直接内存创建和销毁的代价昂贵，但读写性能高（少一次内存复制），适合配合池化功能一起用
- 直接内存对 GC 压力小，因为这部分内存不受 JVM 垃圾回收的管理，但也要注意及时主动释放
3）池化 vs 非池化
池化的最大意义在于可以重用 ByteBuf，优点有
- 没有池化，则每次都得创建新的 ByteBuf 实例，这个操作对直接内存代价昂贵，就算是堆内存，也会增加 GC 压力
- 有了池化，则可以重用池中 ByteBuf 实例，并且采用了与 jemalloc 类似的内存分配算法提升分配效率
- 高并发时，池化功能更节约内存，减少内存溢出的可能
池化功能是否开启，可以通过下面的系统环境变量来设置
-Dio.netty.allocator.type={unpooled|pooled}
- 4.1 以后，非 Android 平台默认启用池化实现，Android 平台启用非池化实现
- 4.1 之前，池化功能还不成熟，默认是非池化实现
4）组成
ByteBuf 由四部分组成
最开始读写指针都在 0 位置
5）写入
方法列表，省略一些不重要的方法
| 方法签名 | 含义 | 备注 |
|---|---|---|
| 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) | 写入字符串 |
注意
- 这些方法的未指明返回值的，其返回值都是 ByteBuf，意味着可以链式调用
- 网络传输，默认习惯是 Big Endian
先写入 4 个字节
buffer.writeBytes(new byte[]{1, 2, 3, 4});
log(buffer);
结果是
read index:0 write index:4 capacity:10
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03 04 |.... |
+--------+-------------------------------------------------+----------------+
再写入一个 int 整数，也是 4 个字节
buffer.writeInt(5);
log(buffer);
结果是
read index:0 write index:8 capacity:10
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03 04 00 00 00 05 |........ |
+--------+-------------------------------------------------+----------------+
还有一类方法是 set 开头的一系列方法，也可以写入数据，但不会改变写指针位置
6）扩容
再写入一个 int 整数时，容量不够了（初始容量是 10），这时会引发扩容
buffer.writeInt(6);
log(buffer);
扩容规则是
- 如何写入后数据大小未超过 512，则选择下一个 16 的整数倍，例如写入后大小为 12 ，则扩容后 capacity 是 16
- 如果写入后数据大小超过 512，则选择下一个 2^n，例如写入后大小为 513，则扩容后 capacity 是 210=1024（29=512 已经不够了）
- 扩容不能超过 max capacity 会报错
结果是
read index:0 write index:12 capacity:16
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03 04 00 00 00 05 00 00 00 06 |............ |
+--------+-------------------------------------------------+----------------+
7）读取
例如读了 4 次，每次一个字节
System.out.println(buffer.readByte());
System.out.println(buffer.readByte());
System.out.println(buffer.readByte());
System.out.println(buffer.readByte());
log(buffer);
读过的内容，就属于废弃部分了，再读只能读那些尚未读取的部分
1
2
3
4
read index:4 write index:12 capacity:16
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 00 00 00 05 00 00 00 06 |........ |
+--------+-------------------------------------------------+----------------+
如果需要重复读取 int 整数 5，怎么办？
可以在 read 前先做个标记 mark
buffer.markReaderIndex();
System.out.println(buffer.readInt());
log(buffer);
结果
5
read index:8 write index:12 capacity:16
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 00 00 00 06 |.... |
+--------+-------------------------------------------------+----------------+
这时要重复读取的话，重置到标记位置 reset
buffer.resetReaderIndex();
log(buffer);
这时
read index:4 write index:12 capacity:16
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 00 00 00 05 00 00 00 06 |........ |
+--------+-------------------------------------------------+----------------+
还有种办法是采用 get 开头的一系列方法，这些方法不会改变 read index
8）retain & release
由于 Netty 中有堆外内存的 ByteBuf 实现，堆外内存最好是手动来释放，而不是等 GC 垃圾回收。
- UnpooledHeapByteBuf 使用的是 JVM 内存，只需等 GC 回收内存即可
- UnpooledDirectByteBuf 使用的就是直接内存了，需要特殊的方法来回收内存
- PooledByteBuf 和它的子类使用了池化机制，需要更复杂的规则来回收内存
回收内存的源码实现，请关注下面方法的不同实现
protected abstract void deallocate()
Netty 这里采用了引用计数法来控制回收内存，每个 ByteBuf 都实现了 ReferenceCounted 接口
- 每个 ByteBuf 对象的初始计数为 1
- 调用 release 方法计数减 1，如果计数为 0，ByteBuf 内存被回收
- 调用 retain 方法计数加 1，表示调用者没用完之前，其它 handler 即使调用了 release 也不会造成回收
- 当计数为 0 时，底层内存会被回收，这时即使 ByteBuf 对象还在，其各个方法均无法正常使用
谁来负责 release 呢？
不是我们想象的（一般情况下）
ByteBuf buf = ...
try {
...
} finally {
buf.release();
}
请思考，因为 pipeline 的存在，一般需要将 ByteBuf 传递给下一个 ChannelHandler，如果在 finally 中 release 了，就失去了传递性（当然，如果在这个 ChannelHandler 内这个 ByteBuf 已完成了它的使命，那么便无须再传递）
基本规则是，谁是最后使用者，谁负责 release，详细分析如下
- 起点，对于 NIO 实现来讲，在 io.netty.channel.nio.AbstractNioByteChannel.NioByteUnsafe#read 方法中首次创建 ByteBuf 放入 pipeline（line 163 pipeline.fireChannelRead(byteBuf)）
- 入站 ByteBuf 处理原则
- 对原始 ByteBuf 不做处理，调用 ctx.fireChannelRead(msg) 向后传递，这时无须 release
- 将原始 ByteBuf 转换为其它类型的 Java 对象，这时 ByteBuf 就没用了，必须 release
- 如果不调用 ctx.fireChannelRead(msg) 向后传递，那么也必须 release
- 注意各种异常，如果 ByteBuf 没有成功传递到下一个 ChannelHandler，必须 release
- 假设消息一直向后传，那么 TailContext 会负责释放未处理消息（原始的 ByteBuf）
- 出站 ByteBuf 处理原则
- 出站消息最终都会转为 ByteBuf 输出，一直向前传，由 HeadContext flush 后 release
- 异常处理原则
- 有时候不清楚 ByteBuf 被引用了多少次，但又必须彻底释放，可以循环调用 release 直到返回 true
TailContext 释放未处理消息逻辑
// io.netty.channel.DefaultChannelPipeline#onUnhandledInboundMessage(java.lang.Object)
protected void onUnhandledInboundMessage(Object msg) {
try {
logger.debug(
"Discarded inbound message {} that reached at the tail of the pipeline. " +
"Please check your pipeline configuration.", msg);
} finally {
ReferenceCountUtil.release(msg);
}
}
具体代码
// io.netty.util.ReferenceCountUtil#release(java.lang.Object)
public static boolean release(Object msg) {
if (msg instanceof ReferenceCounted) {
return ((ReferenceCounted) msg).release();
}
return false;
}
9）slice
【零拷贝】的体现之一，对原始 ByteBuf 进行切片成多个 ByteBuf，切片后的 ByteBuf 并没有发生内存复制，还是使用原始 ByteBuf 的内存，切片后的 ByteBuf 维护独立的 read，write 指针
例，原始 ByteBuf 进行一些初始操作
ByteBuf origin = ByteBufAllocator.DEFAULT.buffer(10);
origin.writeBytes(new byte[]{1, 2, 3, 4});
origin.readByte();
System.out.println(ByteBufUtil.prettyHexDump(origin));
输出
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 02 03 04 |... |
+--------+-------------------------------------------------+----------------+
这时调用 slice 进行切片，无参 slice 是从原始 ByteBuf 的 read index 到 write index 之间的内容进行切片，切片后的 max capacity 被固定为这个区间的大小，因此不能追加 write
ByteBuf slice = origin.slice();
System.out.println(ByteBufUtil.prettyHexDump(slice));
// slice.writeByte(5); 如果执行，会报 IndexOutOfBoundsException 异常
输出
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 02 03 04 |... |
+--------+-------------------------------------------------+----------------+
如果原始 ByteBuf 再次读操作（又读了一个字节）
origin.readByte();
System.out.println(ByteBufUtil.prettyHexDump(origin));
输出
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 03 04 |.. |
+--------+-------------------------------------------------+----------------+
这时的 slice 不受影响，因为它有独立的读写指针
System.out.println(ByteBufUtil.prettyHexDump(slice));
输出
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 02 03 04 |... |
+--------+-------------------------------------------------+----------------+
如果 slice 的内容发生了更改
slice.setByte(2, 5);
System.out.println(ByteBufUtil.prettyHexDump(slice));
输出
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 02 03 05 |... |
+--------+-------------------------------------------------+----------------+
这时，原始 ByteBuf 也会受影响，因为底层都是同一块内存
System.out.println(ByteBufUtil.prettyHexDump(origin));
输出
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 03 05 |.. |
+--------+-------------------------------------------------+----------------+
10）duplicate
【零拷贝】的体现之一，就好比截取了原始 ByteBuf 所有内容，并且没有 max capacity 的限制，也是与原始 ByteBuf 使用同一块底层内存，只是读写指针是独立的
11）copy
会将底层内存数据进行深拷贝，因此无论读写，都与原始 ByteBuf 无关
12）CompositeByteBuf
【零拷贝】的体现之一，可以将多个 ByteBuf 合并为一个逻辑上的 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});
System.out.println(ByteBufUtil.prettyHexDump(buf1));
System.out.println(ByteBufUtil.prettyHexDump(buf2));
输出
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03 04 05 |..... |
+--------+-------------------------------------------------+----------------+
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 06 07 08 09 0a |..... |
+--------+-------------------------------------------------+----------------+
现在需要一个新的 ByteBuf，内容来自于刚才的 buf1 和 buf2，如何实现？
方法1：
ByteBuf buf3 = ByteBufAllocator.DEFAULT
.buffer(buf1.readableBytes()+buf2.readableBytes());
buf3.writeBytes(buf1);
buf3.writeBytes(buf2);
System.out.println(ByteBufUtil.prettyHexDump(buf3));
结果
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03 04 05 06 07 08 09 0a |.......... |
+--------+-------------------------------------------------+----------------+
这种方法好不好？回答是不太好，因为进行了数据的内存复制操作
方法2：
CompositeByteBuf buf3 = ByteBufAllocator.DEFAULT.compositeBuffer();
// true 表示增加新的 ByteBuf 自动递增 write index, 否则 write index 会始终为 0
buf3.addComponents(true, buf1, buf2);
结果是一样的
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03 04 05 06 07 08 09 0a |.......... |
+--------+-------------------------------------------------+----------------+
CompositeByteBuf 是一个组合的 ByteBuf，它内部维护了一个 Component 数组，每个 Component 管理一个 ByteBuf，记录了这个 ByteBuf 相对于整体偏移量等信息，代表着整体中某一段的数据。
- 优点，对外是一个虚拟视图，组合这些 ByteBuf 不会产生内存复制
- 缺点，复杂了很多，多次操作会带来性能的损耗
13）Unpooled
Unpooled 是一个工具类，类如其名，提供了非池化的 ByteBuf 创建、组合、复制等操作
这里仅介绍其跟【零拷贝】相关的 wrappedBuffer 方法，可以用来包装 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 个数超过一个时, 底层使用了 CompositeByteBuf
ByteBuf buf3 = Unpooled.wrappedBuffer(buf1, buf2);
System.out.println(ByteBufUtil.prettyHexDump(buf3));
输出
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03 04 05 06 07 08 09 0a |.......... |
+--------+-------------------------------------------------+----------------+
也可以用来包装普通字节数组，底层也不会有拷贝操作
ByteBuf buf4 = Unpooled.wrappedBuffer(new byte[]{1, 2, 3}, new byte[]{4, 5, 6});
System.out.println(buf4.getClass());
System.out.println(ByteBufUtil.prettyHexDump(buf4));
输出
class io.netty.buffer.CompositeByteBuf
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03 04 05 06 |...... |
+--------+-------------------------------------------------+----------------+
💡 ByteBuf 优势
- 池化 - 可以重用池中 ByteBuf 实例，更节约内存，减少内存溢出的可能
- 读写指针分离，不需要像 ByteBuffer 一样切换读写模式
- 可以自动扩容
- 支持链式调用，使用更流畅
- 很多地方体现零拷贝，例如 slice、duplicate、CompositeByteBuf
4. 双向通信
4.1 练习
实现一个 echo server
编写 server
@Slf4j
public class EchoServer {
public static void main(String[] args) {
NioEventLoopGroup boss = new NioEventLoopGroup();
// 俩个事件循环组作为服务端接受客户端连接和处理客户端连接的线程组
NioEventLoopGroup worker = new NioEventLoopGroup();
new ServerBootstrap()
.group(boss,worker)
.channel((NioServerSocketChannel.class))
.childHandler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) throws Exception {
ch.pipeline().addLast(new StringDecoder());
ch.pipeline().addLast(new StringEncoder());
//读写交给不同的处理器 一个负责读取client发送的消息一个负责写入
ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("服务器接收消息");
log.info((String) msg);
ctx.channel().writeAndFlush(msg);
}
});
//服务器读取数据后 写入通道 触发写入事件
ch.pipeline().addLast(new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.info("出服务器已经发送消息");
super.write(ctx, msg, promise);
}
});
}
}).bind(7878);
}
}
编写 client
/**
* 测试要求
* 1. 编写一个Netty客户端程序，该程序能够发送一个请求并接收一个响应。
* 2. 客户端应该能够连接到指定的服务器，并发送请求数据。
* 3. 服务器应该能够接收请求并处理，然后返回响应数据给客户端。
* 4. 客户端应该能够接收并打印响应数据。
* 5. 客户端和服务器应该能够处理异常情况，并打印错误信息。
*/
@Slf4j
public class InputChannelClient {
public static void main(String[] args) {
NioEventLoopGroup group = new NioEventLoopGroup();
ChannelFuture future = new Bootstrap()
.channel(NioSocketChannel.class)
.group(group)
/**
* 可以简单记忆是这是一个二级的关系
* 通道是双向连接通道,客户端和服务器端都可以进行处理
* 服务器端使用的是ServerSocketChannel 而客户端是SocketChannel
* 服务器端接收到了客户端的连接请求后 childHdander进行处理数据
* 而childhander中处理的是创建连接后的双向NioSocketChannel
* (对等关系)服务端 : 客户端
* ServerSocketChannel -> SocketChanel :SocketChannel
* childhandler : handler
*/
.handler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) throws Exception {
ch.pipeline().addLast(new StringEncoder());
ch.pipeline().addLast(new StringDecoder());
// 日志处理器
// ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("客户端接收消息");
log.info((String) msg);
}
});
}
})
.connect("localhost", 7878);
// 异步监听
future.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
Channel channel = future.channel();
new Thread(()->{
Scanner scanner = new Scanner(System.in);
while (true) {
String line = scanner.nextLine();
if (line.equals("q")){
// channel关闭后 回调函数不在执行 也就是说如果while循环后有逻辑 不会执行
channel.close();//close 也是异步方法 需要考虑异步处理线程的速度
break;
}
else //不是退出操作 写入通道
{
channel.writeAndFlush(line);
}
}
},"input").start();
//获取关闭结果处理
ChannelFuture closeFuture = channel.closeFuture();
//调用关闭channel close api的线程触发该会带哦
closeFuture.addListener((ChannelFutureListener) future1 -> {
log.info("客户端通道关闭");
group.shutdownGracefully();//停止事件循环组继续循环遍历nio线程任务
});
}
});
}
}
似乎没有直接涉及到手动管理或关闭
ByteBuf对象。通常情况下，在Netty中，ByteBuf会被自动释放，而无需手动关闭。Netty的
ByteBuf实现了引用计数机制，当不再有引用指向ByteBuf时，它会被释放。在你的代码中，你主要使用了StringEncoder和StringDecoder处理器，它们应该在内部负责处理ByteBuf的引用计数，而你无需显式地关闭ByteBuf。。
💡 读和写的误解
我最初在认识上有这样的误区，认为只有在 netty，nio 这样的多路复用 IO 模型时，读写才不会相互阻塞，才可以实现高效的双向通信，但实际上，Java Socket 是全双工的：在任意时刻，线路上存在A 到 B 和 B 到 A 的双向信号传输。即使是阻塞 IO，读和写是可以同时进行的，只要分别采用读线程和写线程即可，读不会阻塞写、写也不会阻塞读
例如
public class TestServer {
public static void main(String[] args) throws IOException {
ServerSocket ss = new ServerSocket(8888);
Socket s = ss.accept();
new Thread(() -> {
try {
BufferedReader reader = new BufferedReader(new InputStreamReader(s.getInputStream()));
while (true) {
System.out.println(reader.readLine());
}
} catch (IOException e) {
e.printStackTrace();
}
}).start();
new Thread(() -> {
try {
BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(s.getOutputStream()));
// 例如在这个位置加入 thread 级别断点，可以发现即使不写入数据，也不妨碍前面线程读取客户端数据
for (int i = 0; i < 100; i++) {
writer.write(String.valueOf(i));
writer.newLine();
writer.flush();
}
} catch (IOException e) {
e.printStackTrace();
}
}).start();
}
}
客户端
public class TestClient {
public static void main(String[] args) throws IOException {
Socket s = new Socket("localhost", 8888);
new Thread(() -> {
try {
BufferedReader reader = new BufferedReader(new InputStreamReader(s.getInputStream()));
while (true) {
System.out.println(reader.readLine());
}
} catch (IOException e) {
e.printStackTrace();
}
}).start();
new Thread(() -> {
try {
BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(s.getOutputStream()));
for (int i = 0; i < 100; i++) {
writer.write(String.valueOf(i));
writer.newLine();
writer.flush();
}
} catch (IOException e) {
e.printStackTrace();
}
}).start();
}
}
