技术馆网站建设,北京创意网站建设,在百度上建网站怎么建设,主流建站公司文章目录 二. Netty 入门1. 概述1.1 Netty 是什么#xff1f;1.2 Netty 的作者1.3 Netty 的地位1.4 Netty 的优势 2. Hello World2.1 目标加入依赖 2.2 服务器端2.3 客户端2.4 流程梳理课堂示例服务端客户端 分析提示#xff08;重要#xff09; 3. 组件3.1 EventLoop事件循… 文章目录 二. Netty 入门1. 概述1.1 Netty 是什么1.2 Netty 的作者1.3 Netty 的地位1.4 Netty 的优势 2. Hello World2.1 目标加入依赖 2.2 服务器端2.3 客户端2.4 流程梳理课堂示例服务端客户端 分析提示重要 3. 组件3.1 EventLoop事件循环对象事件循环组课堂示例优雅关闭 演示 NioEventLoop 处理 io 事件笔记示例课堂示例服务端代码客户端代码测试 笔记示例课堂示例服务端代码客户端代码测试 handler 执行中如何实现的换人源码剖析 演示 NioEventLoop 处理普通任务演示 NioEventLoop 处理定时任务 3.2 ChannelChannelFuture笔记示例课堂示例客户端服务端 CloseFuture课堂示例客户端服务端测试 异步提升的是什么 3.3 Future Promise课堂示例TestJdkFutureTestNettyFutureTestNettyPromise 例1例2例3例4例5例6 3.4 Handler Pipeline笔记示例课堂示例服务端客户端测试 EmbeddedChannel使用 3.5 ByteBuf1创建示例 2直接内存 vs 堆内存3池化 vs 非池化4组成5写入6扩容7读取8retain releaseHeadContext TailContext 9slice课堂示例 10duplicate11copy12CompositeByteBuf课堂示例 13Unpooled ByteBuf 优势 4. 双向通信4.1 练习笔记示例编写 server编写 client 课堂示例服务端客户端图示 个人示例服务端客户端 小结作业释放问题 读和写的误解示例服务端客户端 二. Netty 入门
1. 概述
1.1 Netty 是什么
Netty is an asynchronous event-driven network application framework
for rapid development of maintainable high performance protocol servers clients.Netty 是一个异步netty并没有采用异步io这里的异步指的是netty使用了多线程来完成方法的调用和处理结果向分离的、基于事件驱动的网络应用框架用于快速开发可维护、高性能的网络服务器和客户端 补充 调用方法的线程与接收结果的线程是同一个那就意味着阻塞意味着同步
调用方法的线程与接收结果的线程不是同一个那就意味着异步解放了调用方法的线程让调用方法的线程可以处理其它工作。
netty的io模型还是基于多路复用的。
1.2 Netty 的作者
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-afkKSkVO-1690424998728)(assets/0005.png)]
他还是另一个著名网络应用框架 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 ThreadLocalByteBuf ByteBuffer Netty vs 其它网络应用框架 Mina 由 apache 维护将来 3.x 版本可能会有较大重构破坏 API 向下兼容性Netty 的开发迭代更迅速API 更简洁、文档更优秀久经考验16年Netty 版本 2.x 20043.x 20084.x 20135.x 已废弃没有明显的性能提升维护成本高
2. Hello World
2.1 目标
开发一个简单的服务器端和客户端
客户端向服务器端发送 hello, world服务器仅接收不返回
加入依赖
dependencygroupIdio.netty/groupIdartifactIdnetty-all/artifactIdversion4.1.39.Final/version
/dependency2.2 服务器端
new ServerBootstrap().group(new NioEventLoopGroup()) // 1.channel(NioServerSocketChannel.class) // 2.childHandler(new ChannelInitializerNioSocketChannel() { // 3protected void initChannel(NioSocketChannel ch) {ch.pipeline().addLast(new StringDecoder()); // 5ch.pipeline().addLast(new SimpleChannelInboundHandlerString() { // 6Overrideprotected void channelRead0(ChannelHandlerContext ctx, String msg) {System.out.println(msg);}});}}).bind(8080); // 4代码解读 1 处创建 NioEventLoopGroup可以简单理解为 线程池 Selector 后面会详细展开 2 处选择服务 Scoket 实现类其中 NioServerSocketChannel 表示基于 NIO 的服务器端实现其它实现还有 [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-cMNnrJ7N-1690424998731)(assets/0006.png)] 3 处为啥方法叫 childHandler是接下来添加的处理器都是给 SocketChannel 用的而不是给 ServerSocketChannel。ChannelInitializer 处理器仅执行一次它的作用是待客户端 SocketChannel 建立连接后执行 initChannel 以便添加更多的处理器 4 处ServerSocketChannel 绑定的监听端口 5 处SocketChannel 的处理器解码 ByteBuf String 6 处SocketChannel 的业务处理器使用上一个处理器的处理结果
2.3 客户端
new Bootstrap().group(new NioEventLoopGroup()) // 1.channel(NioSocketChannel.class) // 2.handler(new ChannelInitializerChannel() { // 3Overrideprotected 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 的客户端实现其它实现还有 [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-8NEtYj4O-1690424998732)(assets/0007.png)] 3 处添加 SocketChannel 的处理器ChannelInitializer 处理器仅执行一次它的作用是待客户端 SocketChannel 建立连接后执行 initChannel 以便添加更多的处理器 4 处指定要连接的服务器和端口 5 处Netty 中很多方法都是异步的如 connect这时需要使用 sync 方法等待 connect 建立连接完毕 6 处获取 channel 对象它即为通道抽象可以进行数据读写操作 7 处写入消息并清空缓冲区 8 处消息会经过通道 handler 处理这里是将 String ByteBuf 发出 数据经过网络传输到达服务器端服务器端 5 和 6 处的 handler 先后被触发走完一个流程
2.4 流程梳理
课堂示例
服务端
public class HelloServer {public static void main(String[] args) {// 1. 启动器负责组装 netty 组件启动服务器new ServerBootstrap()// 2. BossEventLoop, WorkerEventLoop(selector,thread), group 组// 一个EventLoop就是1个selector 1个thread.group(new NioEventLoopGroup())// 3. 选择 服务器的 ServerSocketChannel 实现.channel(NioServerSocketChannel.class) // OIO BIO// 4. boss 负责处理连接, worker(即child) 负责处理读写此处设置决定了worker(child)能执行哪些操作handler.childHandler(// 5. channel 代表和客户端进行数据读写的通道; Initializer初始化器负责添加别的 handler// 这个初始化器会在客户端连接建立后被调用new ChannelInitializerNioSocketChannel() {Overrideprotected void initChannel(NioSocketChannel ch) throws Exception {// 6. 添加具体 handlerch.pipeline().addLast(new LoggingHandler());ch.pipeline().addLast(new StringDecoder()); // 将 ByteBuf 转换为字符串ch.pipeline().addLast(new ChannelInboundHandlerAdapter() { // 自定义 handlerOverride // 读事件public void channelRead(ChannelHandlerContext ctx,Object msg) throws Exception {System.out.println(msg); // 打印上一步转换好的字符串}});}})// 7. 绑定监听端口.bind(8080);}
}
客户端
public class HelloClient {public static void main(String[] args) throws InterruptedException {// 1. 启动类new Bootstrap()// 2. 添加 EventLoop客户端也可以使用Selector所以这里也指定了eventLoopGroup.group(new NioEventLoopGroup())// 3. 选择客户端 channel 实现.channel(NioSocketChannel.class)// 4. 添加处理器这个初始化器会在连接建立后被调用.handler(new ChannelInitializerNioSocketChannel() {// 在连接建立后被调用Overrideprotected void initChannel(NioSocketChannel ch) throws Exception {ch.pipeline().addLast(new StringEncoder()); // 将字符串编码为ByteBuf}})// 5. 连接到服务器// 发起与服务端建立连接的请求触发accept事件, 服务端有BossEventLoop接收accept建立连接请求,// 当连接建立后, 服务端 和 客户端分别调用初始化方法, 这些初始化方法就把处理器加好了, // 以后再收发数据, 就会经过这些处理器.connect(new InetSocketAddress(localhost, 8080)).sync() // 阻塞方法直到连接建立.channel()// 6. 向服务器发送数据.writeAndFlush(hello, world); // 无论收发数据都会走handler}
}
分析
这个流程初看比较复杂但须把握几个要点当客户端请求与服务端建立连接图中第11步就会触发服务端accept事件由服务端的BossEventLoop处理这个处理器是netty内部的处理的结果就是调用这个连接建立后初始化器的初始化方法—分别调用客户端和服务端的初始化方法初始化方法分别为客户端和服务端的SocketChannel添加好了handler当收发数据的时候就能用上这些handler了比如下面第14步客户端发送数据给服务器会先经过客户端的处理器即StingEncoder将字符串hello转为ByteBuf服务端就会触发read事件接收到ByteBuf交给服务端的处理器依次处理StringDecocder-将ByteBuf转为字符串自定义处理器-将结果输出到控制台
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-as3HuFqm-1690424998733)(assets/0040.png)]
提示重要 一开始需要树立正确的观念 把 channel 理解为数据的通道把 msg 理解为流动的数据最开始输入是 ByteBuf但经过 pipeline 的加工会变成其它类型对象最后输出又变成 ByteBuf把 handler 理解为数据的处理工序 工序有多道合在一起就是 pipelinepipeline 负责发布事件读、读取完成…传播给每个 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 是一组 EventLoopChannel 一般会调用 EventLoopGroup 的 register 方法来绑定其中一个 EventLoop后续这个 Channel 上的 io 事件都由此 EventLoop 来处理保证了 io 事件处理时的线程安全
继承自 netty 自己的 EventExecutorGroup 实现了 Iterable 接口提供遍历 EventLoop 的能力另有 next 方法获取集合中下一个 EventLoop
以一个简单的实现为例
// 内部创建了两个 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.DefaultEventLoop60f82f98
io.netty.channel.DefaultEventLoop35f983a6
io.netty.channel.DefaultEventLoop60f82f98也可以使用 for 循环
DefaultEventLoopGroup group new DefaultEventLoopGroup(2);
for (EventExecutor eventLoop : group) {System.out.println(eventLoop);
}输出
io.netty.channel.DefaultEventLoop60f82f98
io.netty.channel.DefaultEventLoop35f983a6课堂示例
Slf4j
public class TestEventLoop {public static void main(String[] args) {// 1. 创建事件循环组// 如果不传参数默认就是0就会取cpu核心数*2并且最少1个// 可以处理io 事件可以向其提交普通任务还可以向其提交定时任务EventLoopGroup group new NioEventLoopGroup(2); // EventLoopGroup group new DefaultEventLoopGroup(); // 只能处理 普通任务 和 定时任务// 2. 获取下一个事件循环对象// 在2个事件循环对象中轮流着返回System.out.println(group.next()); // io.netty.channel.nio.NioEventLoop553f17cSystem.out.println(group.next()); // io.netty.channel.nio.NioEventLoop4f7d0008System.out.println(group.next()); // io.netty.channel.nio.NioEventLoop553f17cSystem.out.println(group.next()); // io.netty.channel.nio.NioEventLoop4f7d0008// 3. 执行普通任务/*group.next().execute(() - {try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}log.debug(ok);});*/// 4. 执行定时任务group.next().scheduleAtFixedRate(() - {log.debug(ok);}, 0, 1, TimeUnit.SECONDS);log.debug(main);}
}/* 输出如下:
io.netty.channel.nio.NioEventLoop553f17c
io.netty.channel.nio.NioEventLoop4f7d0008
io.netty.channel.nio.NioEventLoop553f17c
io.netty.channel.nio.NioEventLoop4f7d0008
09:53:27 [DEBUG] [main] c.i.n.c.TestEventLoop - main
09:53:27 [DEBUG] [nioEventLoopGroup-2-1] c.i.n.c.TestEventLoop - ok
09:53:28 [DEBUG] [nioEventLoopGroup-2-1] c.i.n.c.TestEventLoop - ok
09:53:29 [DEBUG] [nioEventLoopGroup-2-1] c.i.n.c.TestEventLoop - ok
*/优雅关闭
优雅关闭 shutdownGracefully 方法。该方法会首先切换 EventLoopGroup 到关闭状态从而拒绝新的任务的加入然后在任务队列的任务都处理完成后停止线程的运行。从而确保整体应用是在正常有序的状态下退出的
演示 NioEventLoop 处理 io 事件
笔记示例
服务器端两个 nio worker 工人
new ServerBootstrap().group(new NioEventLoopGroup(1), new NioEventLoopGroup(2)).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializerNioSocketChannel() {Overrideprotected void initChannel(NioSocketChannel ch) {ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {Overridepublic 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 ChannelInitializerNioSocketChannel() {Overrideprotected 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 之间进行了绑定绑定的意思就是说刚开始服务端是哪个eventLoop处理的此客户端后面该客户端发送的消息都由此服务端的此eventLoop处理
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-bi8mZdsT-1690424998734)(assets/0042.png)]
课堂示例
一旦客户端和服务器建立连接channel就会跟一个eventLoop绑定后面该channel的所有读写事件都将有该eventLoop处理
服务端代码
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import lombok.extern.slf4j.Slf4j;import java.net.InetSocketAddress;
import java.nio.charset.Charset;Slf4j
public class EventLoopServer {public static void main(String[] args) {new ServerBootstrap().group(new NioEventLoopGroup()).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializerSocketChannel() {Overrideprotected void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {ByteBuf buf (ByteBuf) msg;log.info(服务端接收到消息: {}, buf.toString(Charset.defaultCharset()));}});}}).bind(new InetSocketAddress(8080));}
}客户端代码
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;import java.net.InetSocketAddress;public class EventLoopClient {public static void main(String[] args) throws Exception {Channel channel new Bootstrap().group(new NioEventLoopGroup()).channel(NioSocketChannel.class).handler(new ChannelInitializerSocketChannel() {Overrideprotected void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new StringEncoder());}}).connect(new InetSocketAddress(8080)).sync().channel();System.out.println(channel);System.in.read(); // 在此处打上debug// 1.idea的debug模式用法// suspend默认会勾选all意思是当运行到断点时其它所有线程都会阻塞。当放开时所有线程都继续运行// 如果勾选thread则只会停止当前线程到该断点,不会阻塞其它线程。// 比如这里打上debug后如果suspend是all那么使用evaluate expression调用// channel.writeAndFlush(..),并不会将此消息发出去,因为发消息的eventloop因为此处的断点而// 被阻塞了。所以要切换suspend为thread这样只会阻塞当前线程不会影响eventloop发消息的线程// 注意下发消息的线程并不是使用channel.writeAndFlush(..)的线程哦。// 2.当这里使用同一个客户端与服务器建立连接后服务器始终会用同一个eventloop来处理该客户端的事件// 可以通过服务端线程日志输出看出来// 结论一旦客户端和服务器建立连接channel就会跟一个eventLoop绑定// 后续的所有请求都会由这同一个eventLoop处理// 一个eventLoop可以绑定多个channel// 一个channel只会绑定给一个eventLoop后面也只会交给此eventLoop处理}
}测试
启动服务端后多次以debug的方式启动客户端注意debug模式的suspend勾选thread让当前线程阻塞时而不会阻塞其它(发送消息的)线程然后分别使用evaluate expression的方式调用channel.writeAndFlush(…)可以发现当使用第一个客户端发送的消息始终是nioEventLoopGroup-2-2这个线程在处理第二个客户端发送的消息始终是nioEventLoopGroup-2-3这个线程在处理。
20:37:04 [INFO ] [nioEventLoopGroup-2-2] c.z.n.EventLoopServer - 服务端接收到消息: aaa
20:37:10 [INFO ] [nioEventLoopGroup-2-2] c.z.n.EventLoopServer - 服务端接收到消息: bbb
20:37:14 [INFO ] [nioEventLoopGroup-2-2] c.z.n.EventLoopServer - 服务端接收到消息: ccc
20:37:30 [INFO ] [nioEventLoopGroup-2-3] c.z.n.EventLoopServer - 服务端接收到消息: 111
20:37:34 [INFO ] [nioEventLoopGroup-2-3] c.z.n.EventLoopServer - 服务端接收到消息: 222
20:37:37 [INFO ] [nioEventLoopGroup-2-3] c.z.n.EventLoopServer - 服务端接收到消息: 333笔记示例
再增加两个非 nio 工人
DefaultEventLoopGroup normalWorkers new DefaultEventLoopGroup(2);
new ServerBootstrap().group(new NioEventLoopGroup(1), new NioEventLoopGroup(2)).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializerNioSocketChannel() {Overrideprotected void initChannel(NioSocketChannel ch) {ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));ch.pipeline().addLast(normalWorkers,myhandler,new ChannelInboundHandlerAdapter() {Overridepublic 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 可以看到nio 工人和 非 nio 工人也分别绑定了 channelLoggingHandler 由 nio 工人执行而我们自己的 handler 由非 nio 工人执行
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-HI0rO5Q3-1690424998734)(assets/0041.png)]
课堂示例
从上面的课堂示例中我们知道了1个channel会绑定worker事件循环组中的1个eventLoop而worker事件循环组的1个eventLoop可以被多个channel绑定那么如果其中1个channel的read事件默认会被绑定的eventLoop处理假设这个read事件耗时很长那么该eventLoop此时势必就无法处理其它channel上的事件所以netty中在添加handler的时候可以自己手动指定事件循环组自己创建的事件循环组那么该handler处理时是交给此指定的事件循环组中的eventLoop处理并且注意此时该客户端也会和这个事件循环组自己创建的事件循环组中的此eventLoop建立绑定关系当该客户端再次触发事件时此handler仍然会交给这个事件循环组中的此eventLoop处理。
服务端代码
Slf4j
public class EventLoopServer {public static void main(String[] args) {// 细分2创建一个独立的 EventLoopGroupDefaultEventLoopGroup group new DefaultEventLoopGroup(); // 只能处理: 普通任何、定时任务new ServerBootstrap()// 2个参数的: 第一个是boss 和 第二个是worker// 细分1boss 只负责 ServerSocketChannel 上 accept 事件;// NioServerSocketChannel只会跟其中的1个eventLoop进行绑定,// 而ServerSocketChannel只有1个因此boss事件循环组中只需要1个线程,// 所以这里不传入参数默认就是1// worker 只负责 socketChannel 上的读写// 这里如果不传入参数默认就是cpu核心数*2这里设置为2意味着有2个eventLoop线程.group(new NioEventLoopGroup(), new NioEventLoopGroup(2)).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializerSocketChannel() {Overrideprotected void initChannel(SocketChannel ch) throws Exception {// 添加1个handler// 此处未指定具体的事件循环组// 因此会用worker事件循环组中建立连接时绑定的EventLoop来处理此handler日志中有体现ch.pipeline().addLast(handler1,new ChannelInboundHandlerAdapter() {Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {ByteBuf buf (ByteBuf) msg;// ByteBuf转字符串, 这里须指定字符集log.info(服务端接收到消息handler1: {},buf.toString(Charset.defaultCharset()));ctx.fireChannelRead(msg); // 让消息传递给下一个handler}})// 再链式调用添加1个handler并且为此handler指定我们上面创建的事件循环组// 那么此handler将会交给指定的事件循环组中的EventLoop所处理并且后面该客户端的channel的// 该handler处理都会交给这个EventLoop所处理因为会绑定到此EventLoop可以从日志中看到// 只要是同一个客户端该handler的处理都是同一个线程来完成的.addLast(group,hander2,new ChannelInboundHandlerAdapter(){Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {ByteBuf buf (ByteBuf) msg;log.info(服务端接收到消息hander2: {}, buf.toString(Charset.defaultCharset()));}});}}).bind(new InetSocketAddress(8080));}客户端代码
分别以debug的方式启动此客户端3次在每次启动的客户端使用evaluate expression调用channel.writeAndFlush(…)发送2次同样的消息以观察效果
public class EventLoopClient {public static void main(String[] args) throws Exception {Channel channel new Bootstrap().group(new NioEventLoopGroup()).channel(NioSocketChannel.class).handler(new ChannelInitializerSocketChannel() {Overrideprotected void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new StringEncoder());}}).connect(new InetSocketAddress(8080)).sync().channel();System.out.println(channel);System.in.read(); // 此处打上debug注意debug模式中的suspend选择thread}
}
测试
可以看到客户端每次发送2次同样的消息过来服务器这边的handler分别是使用不同的线程去处理的并且每个handler都是相同的线程去处理的说明客户端和服务端的handler与eventLoop绑定只要一绑定不管是worker事件循环组中的EventLoop还是自己创建的事件循环组中的eventLoop后面只要是该客户端都会使用已绑定的EventLoop执行此handler
21:47:59 [INFO ] [nioEventLoopGroup-4-1] c.z.n.EventLoopServer - 服务端接收到消息handler1: 1
21:47:59 [INFO ] [defaultEventLoopGroup-2-1] c.z.n.EventLoopServer - 服务端接收到消息hander2: 1
21:48:06 [INFO ] [nioEventLoopGroup-4-1] c.z.n.EventLoopServer - 服务端接收到消息handler1: 1
21:48:06 [INFO ] [defaultEventLoopGroup-2-1] c.z.n.EventLoopServer - 服务端接收到消息hander2: 1
21:48:23 [INFO ] [nioEventLoopGroup-4-2] c.z.n.EventLoopServer - 服务端接收到消息handler1: 2
21:48:23 [INFO ] [defaultEventLoopGroup-2-2] c.z.n.EventLoopServer - 服务端接收到消息hander2: 2
21:48:25 [INFO ] [nioEventLoopGroup-4-2] c.z.n.EventLoopServer - 服务端接收到消息handler1: 2
21:48:25 [INFO ] [defaultEventLoopGroup-2-2] c.z.n.EventLoopServer - 服务端接收到消息hander2: 2
21:48:48 [INFO ] [nioEventLoopGroup-4-1] c.z.n.EventLoopServer - 服务端接收到消息handler1: 3
21:48:48 [INFO ] [defaultEventLoopGroup-2-3] c.z.n.EventLoopServer - 服务端接收到消息hander2: 3
21:48:50 [INFO ] [nioEventLoopGroup-4-1] c.z.n.EventLoopServer - 服务端接收到消息handler1: 3
21:48:50 [INFO ] [defaultEventLoopGroup-2-3] c.z.n.EventLoopServer - 服务端接收到消息hander2: 3handler 执行中如何实现的换人源码剖析
关键代码 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(); // 返回下一个handler的eventLoop这个executor就是eventLoop// 是直接调用if (executor.inEventLoop()) { // 当前 handler 中的线程是否和executor上面这个变量是同一个线程next.invokeChannelRead(m);} // 不是将要执行的代码作为任务提交给下一个事件循环处理换人else {executor.execute(new Runnable() {Overridepublic void run() {next.invokeChannelRead(m);}});}// 结论如果2个handler绑定的是同一个线程eventLoop那么就直接调用。// 如果2个handler绑定的不是同一个线程eventLoop那么就把要调用的代码封装为一个任务对象// 由下一个handler的线程eventLoop来调用
}如果两个 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() 可以用来关闭 channelcloseFuture() 用来处理 channel 的关闭 sync 方法作用是同步等待 channel 关闭而 addListener 方法是异步等待 channel 关闭 pipeline() 方法添加处理器write() 方法将数据写入channel不一定立刻通过网络将数据发出去内部有缓冲机制写入的数据先放入缓冲区当缓冲区满了才会发送过去可以调用flush方法将缓冲区中的数据立刻发送出去writeAndFlush() 方法将数据写入并刷出立刻写入缓冲区并且将缓冲区中的数据发送出去
ChannelFuture
笔记示例
这时刚才的客户端代码
new Bootstrap().group(new NioEventLoopGroup()).channel(NioSocketChannel.class).handler(new ChannelInitializerChannel() {Overrideprotected 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 ChannelInitializerChannel() {Overrideprotected void initChannel(Channel ch) {ch.pipeline().addLast(new StringEncoder());}}).connect(127.0.0.1, 8080); // 1channelFuture.sync().channel().writeAndFlush(new Date() : hello world!);1 处返回的是 ChannelFuture 对象它的作用是利用 channel() 方法来获取 Channel 对象
注意 connect 方法是异步的意味着不等连接建立方法执行就返回了。因此 channelFuture 对象中不能【立刻】获得到正确的 Channel 对象
实验如下
ChannelFuture channelFuture new Bootstrap().group(new NioEventLoopGroup()).channel(NioSocketChannel.class).handler(new ChannelInitializerChannel() {Overrideprotected void initChannel(Channel ch) {ch.pipeline().addLast(new StringEncoder());}}).connect(127.0.0.1, 8080);System.out.println(channelFuture.channel()); // 1
channelFuture.sync(); // 2
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 方法可以让异步操作同步以外还可以使用回调的方式
ChannelFuture channelFuture new Bootstrap().group(new NioEventLoopGroup()).channel(NioSocketChannel.class).handler(new ChannelInitializerChannel() {Overrideprotected 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]
课堂示例
connect命令只是发起了请求建立连接的命令会返回一个channelFuture对象但是此时并未建立连接也就是connect是一个异步非阻塞方法因此此时如果直接在主线程上直接用channelFuture获取的channel然后发送数据是发送不了的。可以调用channelFuture.sync()让主线程阻塞到连接建立后再获取channel在发送消息。或者也可以通过给channelFuture添加一个Listener当连接建立nio线程负责建立连接后由nio线程发送消息
客户端
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;
import lombok.extern.slf4j.Slf4j;import java.net.InetSocketAddress;Slf4j
public class EventLoopClient {public static void main(String[] args) throws Exception {ChannelFuture channelFuture new Bootstrap().group(new NioEventLoopGroup()).channel(NioSocketChannel.class).handler(new ChannelInitializerSocketChannel() {Override // 在连接建立后被调用protected void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new StringEncoder());}})// 1. 连接到服务器// 异步非阻塞, main 发起了调用, 真正执行connect操作的是nio线程// connect是一个异步非阻塞的方法—调用connect方法的线程不关心结果, // 这个线程只负责发起connect方法的调用,// 把这个命令指派给另一个线程真正去做连接操作的是另外的这个线程// 而不是发起connect方法调用的线程其中这个另外的这个线程指的就是// NioEventLoopGroup中的某个线程// 因此当前调用connect方法的线程其实就是main线程并不会在这里阻塞那如果不在这里阻塞的话// 那下面通过channelFuture拿到的channel就是一个还没有建立连接的channel// 那使用这样的channel去发送数据那自然就发送不出去。// 因此下面调用了一个sync方法来保证channel是已经建立好连接了// 再使用channelFuture获取channel这是一种解决方法// netty中有很多的异步非阻塞方法需要注意// netty中带有Future、Promise的类型都是和异步方法配套使用用来处理结果.connect(new InetSocketAddress(8080)); // connect方法返回的是一个ChannelFuture对象// 第一种方式使用sync方法同步处理结果/*// 输出channel-1: [id: 0x0733f705]log.info(channel-1: {}, channelFuture.channel());// 输出channel-2: [id: 0x0733f705, L:/192.168.134.5:50920 - R:0.0.0.0/0.0.0.0:8080]channelFuture.sync(); // 阻塞住当前线程直到nio线程连接建立完毕log.info(channel-2: {}, channelFuture.channel());Channel channel channelFuture.channel();// 2. 向服务器发送数据channel.writeAndFlush(hello,world);*/// 第二种方式使用addListener回调对象方法异步处理结果channelFuture.addListener(new ChannelFutureListener() {// 在nio线程建立连接之后会调用operationComplete方法Overridepublic void operationComplete(ChannelFuture future) throws Exception {Channel channel future.channel();// 输出[nioEventLoopGroup-2-1] c.z.n.EventLoopClient - channel: // [id: 0x8a900cf4, L:/192.168.134.5:51177 - R:0.0.0.0/0.0.0.0:8080]// 注意输出的线程是nio的线程log.info(channel: {},channel);channel.writeAndFlush(hello,world);}});}
}服务端
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.DefaultEventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import lombok.extern.slf4j.Slf4j;import java.net.InetSocketAddress;
import java.nio.charset.Charset;Slf4j
public class EventLoopServer {public static void main(String[] args) {// 细分2创建一个独立的 EventLoopGroupDefaultEventLoopGroup group new DefaultEventLoopGroup(); // 只能处理: 普通任何、定时任务new ServerBootstrap()// 2个参数的: 第一个是boss 和 第二个是worker// 细分1boss 只负责 ServerSocketChannel 上 accept 事件;// NioServerSocketChannel只会跟其中的1个eventLoop进行绑定,// 而ServerSocketChannel只有1个因此boss事件循环组中只需要1个线程,// 所以这里不传入参数默认就是1// worker 只负责 socketChannel 上的读写// 这里如果不传入参数默认就是cpu核心数*2这里设置为2意味着有2个eventLoop线程.group(new NioEventLoopGroup(), new NioEventLoopGroup(2)).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializerSocketChannel() {Overrideprotected void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(handler1,new ChannelInboundHandlerAdapter() {Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {ByteBuf buf (ByteBuf) msg;// ByteBuf转字符串, 这里须指定字符集log.info(服务端接收到消息handler1: {}, buf.toString(Charset.defaultCharset()));ctx.fireChannelRead(msg); // 让消息传递给下一个handler}}).addLast(group,hander2,new ChannelInboundHandlerAdapter(){Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {ByteBuf buf (ByteBuf) msg;log.info(服务端接收到消息hander2: {}, buf.toString(Charset.defaultCharset()));}});}}).bind(new InetSocketAddress(8080));}
}
CloseFuture
Slf4j
public class CloseFutureClient {public static void main(String[] args) throws InterruptedException {NioEventLoopGroup group new NioEventLoopGroup();ChannelFuture channelFuture new Bootstrap().group(group).channel(NioSocketChannel.class).handler(new ChannelInitializerNioSocketChannel() {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({}, channel);new Thread(()-{Scanner scanner new Scanner(System.in);while (true) {String line scanner.nextLine();if (q.equals(line)) {channel.close(); // close 异步操作 1s 之后
// log.debug(处理关闭之后的操作); // 不能在这里善后break;}channel.writeAndFlush(line);}}, input).start();// 获取 CloseFuture 对象 1) 同步处理关闭 2) 异步处理关闭ChannelFuture closeFuture channel.closeFuture();/*log.debug(waiting close...);closeFuture.sync();log.debug(处理关闭之后的操作);*/closeFuture.addListener(new ChannelFutureListener() {Overridepublic void operationComplete(ChannelFuture future) throws Exception {log.debug(处理关闭之后的操作);group.shutdownGracefully();}});}
}课堂示例
现需要在关闭channel前, 做一些善后的工作处理的方式同ChannelFuture。可以往客户端的pipeLine中添加日志handler方便查看netty的运行流程。
客户端
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.logging.LoggingHandler;
import lombok.extern.slf4j.Slf4j;import java.net.InetSocketAddress;
import java.util.Scanner;Slf4j
public class CloseFutureClient {public static void main(String[] args) throws Exception {NioEventLoopGroup group new NioEventLoopGroup();ChannelFuture channelFuture new Bootstrap().group(group).channel(NioSocketChannel.class).handler(new ChannelInitializerSocketChannel() {Overrideprotected void initChannel(SocketChannel ch) throws Exception {// 添加日志处理器方便查看netty的运行流程ch.pipeline().addLast(new LoggingHandler());ch.pipeline().addLast(new StringEncoder());}}).connect(new InetSocketAddress(localhost, 8080));Channel channel channelFuture.sync().channel();log.info(channel已建立好连接: {}, channel);new Thread(()-{Scanner scanner new Scanner(System.in);while (true) {String content scanner.nextLine();if (q.equals(content)) {// 这里只是发送了关闭channel的命令, 是异步非阻塞的方法, // 因此下面关闭操作并不是在channel关闭时执行的channel.close(); log.debug(处理关闭后的操作-2); // 也不能在这里做channel关闭后的操作break;}channel.writeAndFlush(content);}}, input).start();log.debug(处理关闭后的操作-1); // 不能在这里做channel关闭后的操作// 获取closeFuture对象1. 同步处理关闭 2. 异步处理关闭ChannelFuture closeFuture channel.closeFuture();// 1. 同步处理来关闭/*log.debug(enter waiting close...);closeFuture.sync();log.debug(处理关闭后的操作-正确做法1...);*/// 2. 异步处理关闭closeFuture.addListener(new ChannelFutureListener() {Overridepublic void operationComplete(ChannelFuture future) throws Exception {log.debug(处理关闭后的操作-正确做法2...);group.shutdownGracefully(); // 优雅关闭}});}}服务端
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.DefaultEventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import lombok.extern.slf4j.Slf4j;import java.net.InetSocketAddress;
import java.nio.charset.Charset;Slf4j
public class EventLoopServer {public static void main(String[] args) {// 细分2创建一个独立的 EventLoopGroupDefaultEventLoopGroup group new DefaultEventLoopGroup(); // 只能处理: 普通任何、定时任务new ServerBootstrap()// 2个参数的: 第一个是boss 和 第二个是worker// 细分1boss 只负责 ServerSocketChannel 上 accept 事件;// NioServerSocketChannel只会跟其中的1个eventLoop进行绑定,// 而ServerSocketChannel只有1个因此boss事件循环组中只需要1个线程,// 所以这里不传入参数默认就是1// worker 只负责 socketChannel 上的读写// 这里如果不传入参数默认就是cpu核心数*2这里设置为2意味着有2个eventLoop线程.group(new NioEventLoopGroup(), new NioEventLoopGroup(2)).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializerSocketChannel() {Overrideprotected void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(handler1,new ChannelInboundHandlerAdapter() {Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {ByteBuf buf (ByteBuf) msg;// ByteBuf转字符串, 这里须指定字符集log.info(服务端接收到消息handler1: {}, buf.toString(Charset.defaultCharset()));ctx.fireChannelRead(msg); // 让消息传递给下一个handler}}).addLast(group,hander2,new ChannelInboundHandlerAdapter(){Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {ByteBuf buf (ByteBuf) msg;log.info(服务端接收到消息hander2: {}, buf.toString(Charset.defaultCharset()));}});}}).bind(new InetSocketAddress(8080));}
}
测试
启动服务端然后启动客户端先输出aaa然后输入q客户端日志结果如下。注意看下面关闭的过程。
16:18:36 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x7876dd2d] REGISTERED
16:18:36 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x7876dd2d] CONNECT: localhost/127.0.0.1:8080
16:18:36 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x7876dd2d, L:/127.0.0.1:53924 - R:localhost/127.0.0.1:8080] ACTIVE
16:18:36 [INFO ] [main] c.z.n.c.CloseFutureClient - channel已建立好连接: [id: 0x7876dd2d, L:/127.0.0.1:53924 - R:localhost/127.0.0.1:8080]
16:18:36 [DEBUG] [main] c.z.n.c.CloseFutureClient - 处理关闭后的操作-1
aaa
16:18:40 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x7876dd2d, L:/127.0.0.1:53924 - R:localhost/127.0.0.1:8080] WRITE: 3B-------------------------------------------------| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
-------------------------------------------------------------------------
|00000000| 61 61 61 |aaa |
-------------------------------------------------------------------------
16:18:40 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x7876dd2d, L:/127.0.0.1:53924 - R:localhost/127.0.0.1:8080] FLUSH
q
16:18:43 [DEBUG] [input] c.z.n.c.CloseFutureClient - 处理关闭后的操作-2
16:18:43 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x7876dd2d, L:/127.0.0.1:53924 - R:localhost/127.0.0.1:8080] CLOSE
16:18:43 [DEBUG] [nioEventLoopGroup-2-1] c.z.n.c.CloseFutureClient - 处理关闭后的操作-正确做法2...
16:18:43 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x7876dd2d, L:/127.0.0.1:53924 ! R:localhost/127.0.0.1:8080] INACTIVE
16:18:43 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x7876dd2d, L:/127.0.0.1:53924 ! R:localhost/127.0.0.1:8080] UNREGISTERED
异步提升的是什么 有些同学看到这里会有疑问为什么不在一个线程中去执行建立连接、去执行关闭 channel那样不是也可以吗非要用这么复杂的异步方式比如一个线程发起建立连接另一个线程去真正建立连接 还有同学会笼统地回答因为 netty 异步方式用了多线程、多线程就效率高。其实这些认识都比较片面多线程和异步所提升的效率并不是所认为的
结论提升了吞吐量指的是单位时间内处理请求的速度
思考下面的场景4 个医生给人看病每个病人花费 20 分钟而且医生看病的过程中是以病人为单位的一个病人看完了才能看下一个病人。假设病人源源不断地来可以计算一下 4 个医生一天工作 8 小时处理的病人总数是4 * 8 * 3 96
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-vulLSdcF-1690424998736)(assets/0044.png)]
经研究发现看病可以细分为四个步骤经拆分后每个步骤需要 5 分钟如下
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-PBW9dAKZ-1690424998736)(assets/0048.png)]
因此可以做如下优化只有一开始医生 2、3、4 分别要等待 5、10、15 分钟才能执行工作但只要后续病人源源不断地来他们就能够满负荷工作并且处理病人的能力提高到了 4 * 8 * 12 效率几乎是原来的四倍
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-W6XQTFwv-1690424998736)(assets/0047.png)]
要点
单线程没法异步提高效率必须配合多线程、多核 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 Futurenetty FuturePromisecancel取消任务--isCanceled任务是否取消--isDone任务是否完成不能区分成功失败--get获取任务结果阻塞等待--getNow-获取任务结果非阻塞还未产生结果时返回 null-await-等待任务结束如果任务失败不会抛异常而是通过 isSuccess 判断-sync-等待任务结束如果任务失败抛出异常-isSuccess-判断任务是否成功-cause-获取失败信息非阻塞如果没有失败返回null-addLinstener-添加回调异步接收结果-setSuccess--设置成功结果setFailure--设置失败结果
课堂示例
TestJdkFuture
Slf4j
public class TestJdkFuture {public static void main(String[] args) throws ExecutionException, InterruptedException {// 1. 线程池ExecutorService service Executors.newFixedThreadPool(2);// 2. 提交任务FutureInteger future service.submit(new CallableInteger() {Overridepublic Integer call() throws Exception {log.debug(执行计算);Thread.sleep(1000);return 50;}});// 3. 主线程通过 future 来获取结果log.debug(等待结果);log.debug(结果是 {}, future.get());}
}
// 输出
/*
16:27:50 [DEBUG] [pool-1-thread-1] c.i.n.c.TestJdkFuture - 执行计算
16:27:50 [DEBUG] [main] c.i.n.c.TestJdkFuture - 等待结果
16:27:51 [DEBUG] [main] c.i.n.c.TestJdkFuture - 结果是 50
*/TestNettyFuture
Slf4j
public class TestNettyFuture {public static void main(String[] args) throws ExecutionException, InterruptedException {NioEventLoopGroup group new NioEventLoopGroup();EventLoop eventLoop group.next();// 返回的是netty的Future而不是jdk中的Future实际netty的Future继承自jdk中的FutureFutureInteger future eventLoop.submit(new CallableInteger() {Overridepublic Integer call() throws Exception {log.debug(执行计算);Thread.sleep(1000);return 70;}});// 同步方式获取结果jdk中只能使用同步方式获取结果而netty还支持下面的异步方式获取结果// log.debug(等待结果);// log.debug(结果是 {}, future.get());// 异步方式获取结果future.addListener(new GenericFutureListenerFuture? super Integer(){Overridepublic void operationComplete(Future? super Integer future) throws Exception {log.debug(接收结果:{}, future.getNow());}});}
}
// 输出
/*
16:28:19 [DEBUG] [nioEventLoopGroup-2-1] c.i.n.c.TestNettyFuture - 执行计算
16:28:20 [DEBUG] [nioEventLoopGroup-2-1] c.i.n.c.TestNettyFuture - 接收结果:70
*/TestNettyPromise
Slf4j
public class TestNettyPromise {public static void main(String[] args) throws ExecutionException, InterruptedException {// 1. 准备 EventLoop 对象EventLoop eventLoop new NioEventLoopGroup().next();// 2. 可以主动创建 promise, 它就是1个结果容器DefaultPromiseInteger promise new DefaultPromise(eventLoop);new Thread(() - {// 3. 任意一个线程执行计算计算完毕后都可向 promise 填充结果log.debug(开始计算...);try {int i 1 / 1;Thread.sleep(1000);// 设置1个成功的结果promise.setSuccess(80);} catch (Exception e) {e.printStackTrace();promise.setFailure(e);}}).start();// 4. 接收结果的线程log.debug(等待结果...);log.debug(结果是: {}, promise.get());}}// 输出
/*
16:28:45 [DEBUG] [main] c.i.n.c.TestNettyPromise - 等待结果...
16:28:45 [DEBUG] [Thread-0] c.i.n.c.TestNettyPromise - 开始计算...
16:28:46 [DEBUG] [main] c.i.n.c.TestNettyPromise - 结果是: 80
*/例1
同步处理任务成功
DefaultEventLoop eventExecutors new DefaultEventLoop();
DefaultPromiseInteger 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();
DefaultPromiseInteger promise new DefaultPromise(eventExecutors);// 设置回调异步接收结果
promise.addListener(future - {// 这里的 future 就是上面的 promiselog.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();DefaultPromiseInteger 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();
DefaultPromiseInteger 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();
DefaultPromiseInteger 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();
DefaultPromiseInteger 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: DefaultPromise47499c2a(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: DefaultPromise47499c2a(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)
3.4 Handler Pipeline
ChannelHandler 用来处理 Channel 上的各种事件分为入站、出站两种。所有 ChannelHandler 被连成一串就是 Pipeline
入站处理器通常是 ChannelInboundHandlerAdapter 的子类主要用来读取客户端数据写回结果出站处理器通常是 ChannelOutboundHandlerAdapter 的子类主要对写回结果进行加工
打个比喻每个 Channel 是一个产品的加工车间Pipeline 是车间中的流水线ChannelHandler 就是流水线上的各道工序而后面要讲的 ByteBuf 是原材料经过很多工序的加工先经过一道道入站工序再经过一道道出站工序最终变成产品
笔记示例
先搞清楚顺序服务端
new ServerBootstrap().group(new NioEventLoopGroup()).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializerNioSocketChannel() {protected void initChannel(NioSocketChannel ch) {ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) {System.out.println(1);ctx.fireChannelRead(msg); // 1}});ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) {System.out.println(2);ctx.fireChannelRead(msg); // 2}});ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) {System.out.println(3);ctx.channel().write(msg); // 3}});ch.pipeline().addLast(new ChannelOutboundHandlerAdapter(){Overridepublic void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {System.out.println(4);ctx.write(msg, promise); // 4}});ch.pipeline().addLast(new ChannelOutboundHandlerAdapter(){Overridepublic void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {System.out.println(5);ctx.write(msg, promise); // 5}});ch.pipeline().addLast(new ChannelOutboundHandlerAdapter(){Overridepublic 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 ChannelInitializerChannel() {Overrideprotected 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 组成的双向链表
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-j2TQ2WvQ-1690424998737)(assets/0008.png)]
入站处理器中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 触发的原始流程图中数字代表了处理步骤的先后次序
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-rAoPf1dU-1690424998738)(assets/0009.png)]
课堂示例
有3点需要注意
入站处理器的执行顺序是按照添加顺序、出站处理器的执行顺序是与添加顺序正好相反下面的测试只验证了这点ctx.fireChannelRead(msg)的调用会将数据传递给下个 handler如果不调用调用链会断开。并且此方法的参数会作为下一个handler的参数传递下去ctx.writeAndFlush(…)和ch.writeAndFlush(…)有区别ctx.writeAndFlush(…)从当前的handler从后往前找出站处理器 处理当前handler后面的出站处理器被忽略ch.writeAndFlush(…)从tail这个handler从往前找出站处理器 处理
服务端
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;Slf4j
public class TestPipeline {public static void main(String[] args) {new ServerBootstrap().group(new NioEventLoopGroup()).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializerNioSocketChannel() {Overrideprotected void initChannel(NioSocketChannel ch) throws Exception {// 1. 通过 channel 拿到 pipelineChannelPipeline pipeline ch.pipeline();// 2. 添加处理器 head - h1 - h2 - h4 - h3 - h5 - h6 - tail// 其实底层是个双向列表// ChannelInboundHandlerAdapter - 入站处理器只有从channel中读取数据才会触发// ChannelOutboundHandlerAdapter - 出站处理器只有向channel中写入数据才会触发// 注意入站和出战处理器的执行顺序问题入站处理器的执行顺序是按照添加顺序、// 出站处理器的执行顺序是与添加顺序正好相反pipeline.addLast(h1, new ChannelInboundHandlerAdapter(){Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {log.debug(1);ByteBuf buf ((ByteBuf)msg);String name buf.toString(Charset.defaultCharset())super.channelRead(ctx, name); // 将buf转为name传递给下一个handler}});pipeline.addLast(h2, new ChannelInboundHandlerAdapter(){Overridepublic void channelRead(ChannelHandlerContext ctx, Object name) throws Exception {log.debug(2);// 将数据传递给下个 handler如果不调用调用链会断开// 或者调用 ctx.fireChannelRead(new Student(name)); 与下面一样的效果super.channelRead(ctx, name);}});pipeline.addLast(h3, new ChannelInboundHandlerAdapter(){Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {log.debug(3);// 注意下下面2个使用ctx.writeAndFlush(..)和ch.writeAndFlush(..)是有区别// 从当前的handler往前找出站处理器 处理当前handler后面的出站处理器被忽略/*ctx.writeAndFlush(ctx.alloc().buffer().writeBytes(server....getBytes()));*/// 从tail这个handler从后往前找出站处理器 处理/* ch.writeAndFlush(ctx.alloc().buffer().writeBytes(server....getBytes()));*/}});pipeline.addLast(h4, new ChannelOutboundHandlerAdapter(){Overridepublic void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {log.debug(4);super.write(ctx, msg, promise);}});pipeline.addLast(h5, new ChannelOutboundHandlerAdapter(){Overridepublic void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {log.debug(5);super.write(ctx, msg, promise);}});pipeline.addLast(h6, new ChannelOutboundHandlerAdapter(){Overridepublic void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {log.debug(6);super.write(ctx, msg, promise);}});}}).bind(8080);}DataAllArgsConstructorstatic class Student {private String name;}
}客户端
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.logging.LoggingHandler;
import lombok.extern.slf4j.Slf4j;import java.net.InetSocketAddress;
import java.util.Scanner;Slf4j
public class CloseFutureClient {public static void main(String[] args) throws Exception {NioEventLoopGroup group new NioEventLoopGroup();ChannelFuture channelFuture new Bootstrap().group(group).channel(NioSocketChannel.class).handler(new ChannelInitializerSocketChannel() {Overrideprotected void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new LoggingHandler());ch.pipeline().addLast(new StringEncoder());}}).connect(new InetSocketAddress(localhost, 8080));Channel channel channelFuture.sync().channel();log.info(channel已建立好连接: {}, channel);new Thread(()-{Scanner scanner new Scanner(System.in);while (true) {String content scanner.nextLine();if (q.equals(content)) {// 现需要再关闭channel前, 做一些善后的工作channel.close();log.debug(处理关闭后的操作-2); // 也不能在这里做channel关闭后的操作break;}channel.writeAndFlush(content);}}, input).start();log.debug(处理关闭后的操作-1); // 不能在这里做channel关闭后的操作// 获取closeFuture对象1. 同步处理关闭 2. 异步处理关闭ChannelFuture closeFuture channel.closeFuture();// 1. 同步处理来关闭/*log.debug(enter waiting close...);closeFuture.sync();log.debug(处理关闭后的操作-正确做法1...);*/// 2. 异步处理关闭closeFuture.addListener(new ChannelFutureListener() {Overridepublic void operationComplete(ChannelFuture future) throws Exception {log.debug(处理关闭后的操作-正确做法2...);group.shutdownGracefully();}});}}测试
// 服务端日志输出
/*
18:10:26 [DEBUG] [nioEventLoopGroup-2-2] c.z.n.c.TestPipeline - 1
18:10:26 [DEBUG] [nioEventLoopGroup-2-2] c.z.n.c.TestPipeline - 2
18:10:26 [DEBUG] [nioEventLoopGroup-2-2] c.z.n.c.TestPipeline - 3
18:10:26 [DEBUG] [nioEventLoopGroup-2-2] c.z.n.c.TestPipeline - 6
18:10:26 [DEBUG] [nioEventLoopGroup-2-2] c.z.n.c.TestPipeline - 5
18:10:26 [DEBUG] [nioEventLoopGroup-2-2] c.z.n.c.TestPipeline - 4
*/// 客户端日志输出
/*
18:10:23 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x73322c72] REGISTERED
18:10:23 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x73322c72] CONNECT: localhost/127.0.0.1:8080
18:10:23 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x73322c72, L:/127.0.0.1:55902 - R:localhost/127.0.0.1:8080] ACTIVE
18:10:23 [INFO ] [main] c.z.n.c.CloseFutureClient - channel已建立好连接: [id: 0x73322c72, L:/127.0.0.1:55902 - R:localhost/127.0.0.1:8080]
18:10:23 [DEBUG] [main] c.z.n.c.CloseFutureClient - 处理关闭后的操作-1
aaa
18:10:26 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x73322c72, L:/127.0.0.1:55902 - R:localhost/127.0.0.1:8080] WRITE: 3B-------------------------------------------------| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
-------------------------------------------------------------------------
|00000000| 61 61 61 |aaa |
-------------------------------------------------------------------------
18:10:26 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x73322c72, L:/127.0.0.1:55902 - R:localhost/127.0.0.1:8080] FLUSH
18:10:26 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x73322c72, L:/127.0.0.1:55902 - R:localhost/127.0.0.1:8080] READ: 9B-------------------------------------------------| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
-------------------------------------------------------------------------
|00000000| 73 65 72 76 65 72 2e 2e 2e |server... |
-------------------------------------------------------------------------
18:10:26 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x73322c72, L:/127.0.0.1:55902 - R:localhost/127.0.0.1:8080] READ COMPLETE*/EmbeddedChannel使用
可以方便快捷测试
Slf4j
public class TestEmbeddedChannel {public static void main(String[] args) {ChannelInboundHandlerAdapter h1 new ChannelInboundHandlerAdapter() {Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {log.debug(1);super.channelRead(ctx, msg);}};ChannelInboundHandlerAdapter h2 new ChannelInboundHandlerAdapter() {Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {log.debug(2);super.channelRead(ctx, msg);}};ChannelOutboundHandlerAdapter h3 new ChannelOutboundHandlerAdapter() {Overridepublic void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {log.debug(3);super.write(ctx, msg, promise);}};ChannelOutboundHandlerAdapter h4 new ChannelOutboundHandlerAdapter() {Overridepublic 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()));// 模拟出站操作channel.writeOutbound(ByteBufAllocator.DEFAULT.buffer().writeBytes(world.getBytes()));}
}// 输出
/*
18:23:48 [DEBUG] [main] c.i.n.c.TestEmbeddedChannel - 1
18:23:48 [DEBUG] [main] c.i.n.c.TestEmbeddedChannel - 2
18:23:48 [DEBUG] [main] c.i.n.c.TestEmbeddedChannel - 4
18:23:48 [DEBUG] [main] c.i.n.c.TestEmbeddedChannel - 3
*/3.5 ByteBuf
是对字节数据的封装
1创建
ByteBuf buffer ByteBufAllocator.DEFAULT.buffer(10);
log(buffer);上面代码创建了一个默认的 ByteBuf池化基于直接内存的 ByteBuf初始容量是 10
输出
read index:0 write index:0 capacity:10其中 log 方法参考如下方便查看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());
}示例
public class TestByteBuf {public static void main(String[] args) {// 如果未传入参数, 默认容量为256; 可指定大小当写入的字节数量超过设置的容量大小的话, ByteBuf会自动扩容ByteBuf buf ByteBufAllocator.DEFAULT.buffer();System.out.println(buf.getClass()); // class io.netty.buffer.PooledUnsafeDirectByteBufSystem.out.println(buf.maxCapacity()); // 2147483647// 未写入数据之前, ByteBuf的capacity默认为256log(buf); StringBuilder sb new StringBuilder();for (int i 0; i 32; i) {sb.append(a);}// 写入字节数组buf.writeBytes(sb.toString().getBytes());log(buf);}public 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());}
}// 输出如下
/*
class io.netty.buffer.PooledUnsafeDirectByteBuf
2147483647
read index:0 write index:0 capacity:256read index:0 write index:32 capacity:256-------------------------------------------------| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
-------------------------------------------------------------------------
|00000000| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000010| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
-------------------------------------------------------------------------
*/2直接内存 vs 堆内存
netty默认情况下都使用的是直接内存作为ByteBuf的内存。
可以使用下面的代码来明确创建池化基于堆的 ByteBuf
ByteBuf buffer ByteBufAllocator.DEFAULT.heapBuffer(10);也可以使用下面的代码来明确创建池化基于直接内存的 ByteBuf调用buffer()方法返回的是直接内存
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 之前池化功能还不成熟默认是非池化实现
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-LTSCrKgZ-1690424998739)(assets/image-20230726205205841.png)]
4组成
ByteBuf 由四部分组成
最大容量默认为Integer.MAX_VALUE即2147483647
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-5evff23P-1690424998739)(assets/0010.png)]
最开始读写指针都在 0 位置 与jdk的ByteBuffer相比 jdk中的ByteBuffer读写共用1个指针如果要读数据须切换到读模式如果要写须调用clear或compact切换到写模式
5写入
方法列表省略一些不重要的方法
方法签名含义备注writeBoolean(boolean value)写入 boolean 值用一字节 01|00 代表 true|falsewriteByte(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 的 ByteBufwriteBytes(byte[] src)写入 byte[]writeBytes(ByteBuffer src)写入 nio 的 ByteBufferint 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 是 210102429512 已经不够了扩容不能超过 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);读过的内容就属于废弃部分了再读只能读那些尚未读取的部分注意看调用log方法打印buf是不会打印已经读取完的数据
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
8retain release
由于 Netty 中有堆外内存的 ByteBuf 实现堆外内存最好是手动来释放而不是等 GC 垃圾回收。
UnpooledHeapByteBuf 使用的是 JVM 内存只需等 GC 回收内存即可UnpooledDirectByteBuf 使用的就是直接内存了需要特殊的方法来回收内存PooledByteBuf 和它的子类使用了池化机制需要更复杂的规则来回收内存 回收内存的源码实现请关注下面方法的不同实现 protected abstract void deallocate() Netty 这里采用了引用计数法来控制回收内存每个 ByteBuf 都实现了 ReferenceCounted 接口
每个 ByteBuf 对象的初始计数为 1调用 release 方法计数减 1如果计数为 0ByteBuf 内存被回收调用 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 放入 pipelineline 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 释放未处理消息逻辑TailContext实现了ChannelInboundHandler因此会处理入站消息
// 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;
}HeadContext TailContext
HeadContext既实现了ChannelInboundHandler接口也实现了ChannelOutboundHandler接口在出站的时候有对ByteBuf作引用计数的释放
TailContext只实现了ChannelInboundHandler在入站的时候有对ByteBuf作引用计数的释放
9slice
【零拷贝】的体现之一对原始 ByteBuf 进行切片成多个 ByteBuf切片后的 ByteBuf 并没有发生内存复制还是使用原始 ByteBuf 的内存切片后的 ByteBuf 维护独立的 readwrite 指针
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-CYNrAoVi-1690424998740)(assets/0011.png)]
例原始 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 |.. |
-------------------------------------------------------------------------课堂示例
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import lombok.extern.slf4j.Slf4j;import static io.netty.buffer.ByteBufUtil.appendPrettyHexDump;
import static io.netty.util.internal.StringUtil.NEWLINE;Slf4j
public class TestSlice {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});// 切片过程中, 没有发生数据复制ByteBuf bufSlice1 buf.slice(0, 5);// bufSlice1.retain();ByteBuf bufSlice2 buf.slice(5, 5);// bufSlice2.retain();log(buf);log(bufSlice1);log(bufSlice2);// 此处会报错, 切片后的 max capacity 被固定为这个区间的大小因此不能追加 write// bufSlice1.writeByte(x);log.info();buf.setByte(0, z);log(buf);log(bufSlice1);buf.release(); // 释放原始的buflog(bufSlice1); // 上面将原始的buf给释放了这里再使用切片的buf就会报错// 如果要让这里不报错可以在前面切完片之后, 调用bufSlice1.retain();// bufSlice1.release(); // 用完了切片之后, 释放掉// bufSlice2.release(); // 用完了切片之后, 释放掉}// 注意这个方法只会打印读指针和写指针之间的可读部分的数据已经读过的字节属于废弃部分不会打印出来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());}}
// 输出
/*
read index:0 write index:10 capacity:10-------------------------------------------------| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
-------------------------------------------------------------------------
|00000000| 61 62 63 64 65 66 67 68 69 6a |abcdefghij |
-------------------------------------------------------------------------
read index:0 write index:5 capacity:5-------------------------------------------------| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
-------------------------------------------------------------------------
|00000000| 61 62 63 64 65 |abcde |
-------------------------------------------------------------------------
read index:0 write index:5 capacity:5-------------------------------------------------| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
-------------------------------------------------------------------------
|00000000| 66 67 68 69 6a |fghij |
-------------------------------------------------------------------------
23:06:47 [INFO ] [main] c.z.n.b.TestSlice -
read index:0 write index:10 capacity:10-------------------------------------------------| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
-------------------------------------------------------------------------
|00000000| 7a 62 63 64 65 66 67 68 69 6a |zbcdefghij |
-------------------------------------------------------------------------
read index:0 write index:5 capacity:5-------------------------------------------------| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
-------------------------------------------------------------------------
|00000000| 7a 62 63 64 65 |zbcde |
-------------------------------------------------------------------------
Exception in thread main io.netty.util.IllegalReferenceCountException: refCnt: 0at io.netty.buffer.AbstractByteBuf.ensureAccessible(AbstractByteBuf.java:1464)at io.netty.buffer.AbstractByteBuf.checkIndex(AbstractByteBuf.java:1388)at io.netty.buffer.AbstractByteBuf.checkIndex(AbstractByteBuf.java:1384)at io.netty.buffer.AbstractByteBuf.getByte(AbstractByteBuf.java:361)at io.netty.buffer.AbstractUnpooledSlicedByteBuf.getByte(AbstractUnpooledSlicedByteBuf.java:120)at io.netty.buffer.AbstractByteBuf.getUnsignedByte(AbstractByteBuf.java:374)at io.netty.buffer.ByteBufUtil$HexUtil.appendPrettyHexDump(ByteBufUtil.java:1143)at io.netty.buffer.ByteBufUtil$HexUtil.access$300(ByteBufUtil.java:982)at io.netty.buffer.ByteBufUtil.appendPrettyHexDump(ByteBufUtil.java:978)at io.netty.buffer.ByteBufUtil.appendPrettyHexDump(ByteBufUtil.java:969)at com.zzhua.netty.byteBuf.TestSlice.log(TestSlice.java:59)at com.zzhua.netty.byteBuf.TestSlice.main(TestSlice.java:40)
*/10duplicate
【零拷贝】的体现之一就好比截取了原始 ByteBuf 所有内容并且没有 max capacity 的限制也是与原始 ByteBuf 使用同一块底层内存只是读写指针是独立的
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-JIqLghr1-1690424998741)(assets/0012.png)]
11copy
会将底层内存数据进行深拷贝因此无论读写都与原始 ByteBuf 无关
12CompositeByteBuf
【零拷贝】的体现之一可以将多个 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 不会产生内存复制缺点复杂了很多多次操作会带来性能的损耗
课堂示例
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.CompositeByteBuf;import static cn.itcast.netty.c4.TestByteBuf.log;public class TestCompositeByteBuf {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});/*// 这样操作会发生数据复制ByteBuf buffer ByteBufAllocator.DEFAULT.buffer();buffer.writeBytes(buf1).writeBytes(buf2);log(buffer);*/CompositeByteBuf buffer ByteBufAllocator.DEFAULT.compositeBuffer();// 设置为true,可以将写指针移动到后面; 如果不设置为true,写指针还是在0的位置buffer.addComponents(true, buf1, buf2);log(buffer);// 证明没有发生数据复制buf1.setByte(0, 2);log(buffer);// 注意还是要注意release的问题为了避免引用计数意外的减成了0可以调用ByteBuf#retain()方法}
}13Unpooled
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
new ServerBootstrap().group(new NioEventLoopGroup()).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializerNioSocketChannel() {Overrideprotected void initChannel(NioSocketChannel ch) {ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) {ByteBuf buffer (ByteBuf) msg;System.out.println(buffer.toString(Charset.defaultCharset()));// 建议使用 ctx.alloc() 创建 ByteBufByteBuf response ctx.alloc().buffer();response.writeBytes(buffer);ctx.writeAndFlush(response); // 写出的ByteBuf// 思考需要释放 buffer 吗 我觉得这里需要释放因为这里没有经过TaiContext// 思考需要释放 response 吗我觉得这里不需要释放因为HeadContext会释放它}});}}).bind(8080);编写 client
NioEventLoopGroup group new NioEventLoopGroup();
Channel channel new Bootstrap().group(group).channel(NioSocketChannel.class).handler(new ChannelInitializerNioSocketChannel() {Overrideprotected void initChannel(NioSocketChannel ch) throws Exception {ch.pipeline().addLast(new StringEncoder());ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {// 此方法会在连接建立后触发Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {// 建议使用ctx.alloc() 创建bufferByteBuf buf ctx.alloc().buffer();// 首次建立连接, 发送hello消息buf.writeBytes(hello server, Im a client.getBytes());ctx.writeAndFlush(buf);// 思考需要释放buffer吗 我觉得不需要释放因为会经过HeadContext}Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) {ByteBuf buffer (ByteBuf) msg;System.out.println(buffer.toString(Charset.defaultCharset()));// 思考需要释放 buffer 吗我觉得需要释放因为这里没有经过TaiContext}});}}).connect(127.0.0.1, 8080).sync().channel();channel.closeFuture().addListener(future - {group.shutdownGracefully();
});new Thread(() - {Scanner scanner new Scanner(System.in);while (true) {String line scanner.nextLine();if (q.equals(line)) {channel.close();break;}channel.writeAndFlush(line);}
}).start();课堂示例
服务端客户端图示
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-QogKTEYm-1690424998742)(assets/image-20230727101843855.png)]
个人示例
服务端
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringEncoder;
import lombok.extern.slf4j.Slf4j;import java.net.InetSocketAddress;
import java.nio.charset.Charset;Slf4j
public class EchoServer {public static void main(String[] args) {new ServerBootstrap().group(new NioEventLoopGroup()).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializerSocketChannel() {Overrideprotected void initChannel(SocketChannel ch) throws Exception {// ch.pipeline().addLast(new LoggingHandler());ch.pipeline().addLast(new StringEncoder());ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {ByteBuf buf (ByteBuf) msg;String content buf.toString(Charset.defaultCharset());log.info(服务端收到客户端的消息: {}, content);ctx.writeAndFlush(content);ctx.fireChannelRead(buf); // 交给TailContext去释放这里不需要释放}});}}).bind(new InetSocketAddress(8080));}}客户端
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import lombok.extern.slf4j.Slf4j;import java.net.InetSocketAddress;
import java.util.Scanner;Slf4j
public class EchoClient {public static void main(String[] args) throws InterruptedException {NioEventLoopGroup group new NioEventLoopGroup();ChannelFuture channelFuture new Bootstrap().group(group).channel(NioSocketChannel.class).handler(new ChannelInitializerSocketChannel() {Overrideprotected void initChannel(SocketChannel ch) throws Exception {// ch.pipeline().addLast(new LoggingHandler());ch.pipeline().addLast(new StringEncoder());ch.pipeline().addLast(new StringDecoder());ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {// 此方法会在连接建立后触发Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {ByteBuf buf ctx.alloc().buffer();buf.writeBytes(hello server, Im a client.getBytes());ctx.writeAndFlush(buf); // 由HeadContext释放这里不需要释放}Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {log.info(客户端收到服务端消息: {}, msg);// 疑惑我这里使用了 StringDecoder这种写法要不要释放? // 查看了StringDecoder的源码它在finally里面就把ByteBuf给释放了super.channelRead(ctx, msg); }});}}).connect(new InetSocketAddress(8080));Channel channel channelFuture.sync().channel();Scanner scanner new Scanner(System.in);while (true) {System.out.println(请输入消息: );String content scanner.nextLine();if (!q.equals(content)) {channel.writeAndFlush(content);} else {channel.closeFuture().addListener(new ChannelFutureListener() {Overridepublic void operationComplete(ChannelFuture future) throws Exception {group.shutdownGracefully();}});channel.close();break;}}}}小结作业释放问题
在 Server 中由于buffer是池化的直接内存不受 JVM 垃圾回收的管理要注意及时主动释放并且并没有后序传递所以需要释放。 而 writeAndFlush 调用了出站处理器response会最终会到tailX所以不用释放。但是ctx 和 channel 的 writeAndFlush 不同ctx 是不会到 tail 的只会向前找出站处理器到达 head然后由 head 释放因此还是不用自己释放。 在 Client 中当前Handler是最后一个拿到buf的所以需要释放。 读和写的误解
我最初在认识上有这样的误区认为只有在 nettynio 这样的多路复用 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();}
}
