-
Netty 入门
黑马程序员Netty笔记合集
注意：由于章节连贯，此套笔记更适合学习《黑马Netty全套课程》的同学参考、复习使用。一、概述
1.1 Netty 是什么？
Netty is an asynchronous event-driven network application framework for rapid development of maintainable high performance protocol servers & clients.- 1
- 2
Netty 是一个异步的、基于事件驱动的网络应用框架，用于快速开发可维护、高性能的网络服务器和客户端
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 已废弃（没有明显的性能提升，维护成本高）
💡 异步提升的是什么？
单线程没法异步提高效率，必须配合多线程、多核 cpu 才能发挥异步的优势
- 要点1：异步提高的是吞吐量。
- 要点2：异步没有缩短响应时间，反而有所增加
- 要点3：合理进行任务拆分，也是利用异步的关键
二、Hello World
2.1 目标
开发一个简单的服务器端和客户端
- 客户端向服务器端发送 hello, world
- 服务器仅接收，不返回
加入依赖
<dependency> <groupId>io.nettygroupId> <artifactId>netty-allartifactId> <version>4.1.39.Finalversion> dependency>- 1
- 2
- 3
- 4
- 5
2.2 服务器端
new ServerBootstrap() //1 .channel(NioServerSocketChannel.class) //2 .group(new NioEventLoopGroup()) //3 .childHandler( //4 new ChannelInitializer<NioSocketChannel>() { //5 @Override protected void initChannel(NioSocketChannel ch) throws Exception { //6 //SocketChannel 处理器：解码 ByteBuf => String ch.pipeline().addLast(new StringDecoder()); //业务处理器：使用上一个处理器的处理结果 ch.pipeline().addLast(new SimpleChannelInboundHandler<String>() { //读事件 @Override protected void channelRead0(ChannelHandlerContext channelHandlerContext, String msg) throws Exception { System.out.println(msg); } }); } }) .bind(8080); //7- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
-
启动器：负责组装 netty 组件，启动服务器
-
选择服务器的 ServerSocketChannel 实现类。 NioServerSocketChannel 表示基于 NIO 的服务器端实现
-
创建 Selector 线程处理组：简单理解为 线程池+Selector。[BossEventLoop、WorkerEventLoop(selector,thread)]
-
SocketChannel 连接处理器：决定了worker能执行哪些操作
-
初始化器：待客户端 SocketChannel 建立连接后，执行 initChannel 以便添加更多的处理器。仅执行一次。
-
添加具体 handler：处理 SocketChannel
-
绑定监听端口
2.3 客户端
new Bootstrap() //1 .channel(NioSocketChannel.class) //2 .group(new NioEventLoopGroup()) //3 .handler(new ChannelInitializer<Channel>() { //4 @Override //建立连接后调用 protected void initChannel(Channel ch) throws Exception { //消息会经过通道 handler 处理：将 String => ByteBuf 发出 ch.pipeline().addLast(new StringEncoder()); } }) .connect("127.0.0.1",8080) //5 .sync() //6 .channel() //7 .writeAndFlush(new Date()+":Hello World!"); //8- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
-
启动类
-
选择客户 Channel 实现类。NioSocketChannel 表示基于 NIO 的客户端实现
-
创建 Selector 线程处理组
-
添加 SocketChannel 的处理器：待客户端 SocketChannel 建立连接后，执行 initChannel 以添加更多的处理器。ChannelInitializer 仅执行一次。
-
指定连接的服务器和端口
-
同步操作：阻塞等待Netty中异步方法的完成，如等待 connect 建立连接完毕
-
获取 channel 对象：它即为通道抽象，可以进行数据读写操作
-
写入消息并清空缓冲区
2.4 流程梳理
2.5 总结💡
- channel：数据的通道
- msg：流动的数据。最开始输入是 ByteBuf，但经过 pipeline(流水线) 的加工，会变成其它类型对象，最后输出又变成 ByteBuf
- handler：数据的处理工序。Inbound（入栈）、Outbound（出栈）
- 对自己感兴趣的事件进行处理（重写了相应事件处理方法）
- pipeline：多道工序的组合。
- 负责发布事件（读、读取完成…）传播给每个 handler
- eventLoop ：处理数据的工人，管理多个 channel 的 io 操作
- 一旦负责了某个 channel，就要负责到底（绑定）
- 工人既可以执行 io 操作，也可以进行任务处理。每位工人有任务队列，队列里可以堆放多个 channel 的待处理任务，任务分为普通任务、定时任务
- 工人按照 pipeline 顺序，依次按照 handler 的规划（代码）处理数据，可以为每道工序指定不同的工人
三、组件
3.1 EventLoop
3.1.1 介绍
EventLoop：事件循环对象
EventLoop 本质是一个单线程执行器（同时维护了一个 Selector），里面有 run 方法处理 Channel 上源源不断的 io 事件。
它的继承关系比较复杂
- 一条线是继承 …netty…EventLoopGroup—>…netty…EventExecutorGroup—> j.u.c.ScheduledExecutorService 因此包含了线程池中所有的方法
- 另一条线是继承 …netty…OrderedEventExecutor—>…netty…EventExecutor
- 提供了 boolean inEventLoop(Thread thread) 方法判断一个线程是否属于此 EventLoop
- 提供了 parent 方法来看看自己属于哪个 EventLoopGroup
EventLoopGroup：事件循环组
EventLoopGroup 是一组 EventLoop，Channel 一般会调用 EventLoopGroup 的 register 方法来绑定其中一个 EventLoop，后续这个 Channel 上的 io 事件都由此 EventLoop 来处理（保证了 io 事件处理时的线程安全）
- 没有指定线程数时：max（1，SystemPropertyUtil.getInt(“io.netty.eventLoopThreads”,NettyRuntime.availableProcessors()*2))
- 继承自 …netty…EventExecutorGroup
- 实现了 Iterable 接口提供遍历 EventLoop 的能力
- 另有 next 方法获取集合中下一个 EventLoop
3.1.2 普通、定时事件
public static void main(String[] args) { //1.创建事件循环组 //没有指定线程数：max（1，SystemPropertyUtil.getInt("io.netty.eventLoopThreads",NettyRuntime.availableProcessors()*2)) EventLoopGroup group=new NioEventLoopGroup(2); //2.获取下一个事件循环对象 log.debug(group.next().toString()); log.debug(group.next().toString()); log.debug(group.next().toString()); log.debug(group.next().toString()); //3.执行普通事件 group.execute(()->{ try { Thread.sleep(1000L); } catch (InterruptedException e) { e.printStackTrace(); } log.debug("事件循环对象执行-普通任务"); }); //4.执行定时任务：0秒后开始执行，间隔1秒再次执行 group.scheduleAtFixedRate(()->{ log.debug("事件循环对象执行-定时任务"); },2,2, TimeUnit.SECONDS); log.debug("主线程"); }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
10:43:13 [DEBUG] [main] o.e.j.e.EventLoopTest : io.netty.channel.nio.NioEventLoop@3bfdc050 10:43:13 [DEBUG] [main] o.e.j.e.EventLoopTest : io.netty.channel.nio.NioEventLoop@1bce4f0a 10:43:13 [DEBUG] [main] o.e.j.e.EventLoopTest : io.netty.channel.nio.NioEventLoop@3bfdc050 10:43:13 [DEBUG] [main] o.e.j.e.EventLoopTest : io.netty.channel.nio.NioEventLoop@1bce4f0a 10:43:13 [DEBUG] [main] o.e.j.e.EventLoopTest : 主线程 10:43:14 [DEBUG] [nioEventLoopGroup-2-1] o.e.j.e.EventLoopTest : 事件循环对象执行-普通任务 10:43:15 [DEBUG] [nioEventLoopGroup-2-2] o.e.j.e.EventLoopTest : 事件循环对象执行-定时任务 10:43:17 [DEBUG] [nioEventLoopGroup-2-2] o.e.j.e.EventLoopTest : 事件循环对象执行-定时任务- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
3.1.3 IO 事件
细分1：两个事件循环组分别处理 accept事件 和 读写事件
public static void main(String[] args) { new ServerBootstrap() //细分一：将只负责accept事件的 Boss 和 只处理读写事件的 woker 分成两个事件循环组 .group(new NioEventLoopGroup(),new NioEventLoopGroup(2)) .channel(NioServerSocketChannel.class) .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){ @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { ByteBuf buf= (ByteBuf) msg; log.debug(buf.toString(Charset.forName("UTF-8"))); } }); } }) .bind(8080); }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
/** * 客户端 */ public static void main(String[] args) throws InterruptedException { Channel channel = new Bootstrap() .group(new NioEventLoopGroup()) .channel(NioSocketChannel.class) .handler(new ChannelInitializer<NioSocketChannel>() { @Override protected void initChannel(NioSocketChannel ch) throws Exception { ch.pipeline().addLast(new StringEncoder()); } }) .connect(new InetSocketAddress("127.0.0.1", 8080)) .sync() .channel(); System.out.println(channel); System.out.println(""); }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 两个工人轮流处理
细分2：增加处理 非IO事件 的事件循环组对象，让 IO循环组对象 可以及时处理多个客户端的IO请求
/** * 服务端 */ public static void main(String[] args) { //细分二：增加处理 非IO事件 的事件循环组对象。让 IO循环组对象 可以及时处理多个客户端的IO请求 DefaultEventLoopGroup defaultGroup = new DefaultEventLoopGroup(); new ServerBootstrap() //细分一：将只负责accept事件的 Boss 和 只处理读写事件的 woker 分成两个事件循环组 .group(new NioEventLoopGroup(),new NioEventLoopGroup(2)) .channel(NioServerSocketChannel.class) .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { //IO循环组对象 处理 ch.pipeline().addLast("handler1",new ChannelInboundHandlerAdapter(){ @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { ByteBuf buf= (ByteBuf) msg; log.debug(buf.toString(Charset.forName("UTF-8"))); ctx.fireChannelRead(msg); // 让消息传递给下一个handler } }); //非IO事件循环组对象 处理 ch.pipeline().addLast(defaultGroup,"handler2",new ChannelInboundHandlerAdapter(){ @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { ByteBuf buf= (ByteBuf) msg; log.debug(buf.toString(Charset.forName("UTF-8"))); } }); } }) .bind(8080); }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
/** * 客户端 */ public static void main(String[] args) throws InterruptedException { Channel channel = new Bootstrap() .group(new NioEventLoopGroup()) .channel(NioSocketChannel.class) .handler(new ChannelInitializer<NioSocketChannel>() { @Override protected void initChannel(NioSocketChannel ch) throws Exception { ch.pipeline().addLast(new StringEncoder()); } }) .connect(new InetSocketAddress("127.0.0.1", 8080)) .sync() .channel(); System.out.println(channel); System.out.println(""); }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 客户端的请求被不同的工人处理：让处理IO事件的工人有更多时间处理其他客户端的请求
💡 源码分析: 线程变更
- 如果当前工序与下一道工序是同一个工人（同一线程）处理，让工人继续处理
- 否则，将 下一道工序的代码 作为任务提交给 另一个工人（另一个线程） 处理，即 线程变更
/** * 关键代码 io.netty.channel.AbstractChannelHandlerContext#invokeChannelRead() */ static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) { final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next); // 1.获取下一个 hander 的事件循环对象 EventExecutor executor = next.executor(); // 2.判断：下一道工序的处理工人 与 当前工人 是否是同一个？ if (executor.inEventLoop()) { //3.是，让该工人继续处理 next.invokeChannelRead(m); } // 4.不是，将 下一道工序的代码 作为任务提交给 另一个工人 处理（换人） else { executor.execute(new Runnable() { @Override public void run() { next.invokeChannelRead(m); } }); } }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
💡 优雅关闭
优雅关闭
shutdownGracefully方法。该方法会首先切换EventLoopGroup到关闭状态从而拒绝新的任务的加入，然后在任务队列的任务都处理完成后，停止线程的运行。从而确保整体应用是在正常有序的状态下退出的3.2 Channel
3.2.1 ChannelFuture
- sync()：同步等待 通道内的事件完成
- channel()：获取与服务器的传输通道
- addListener(GenericFutureListener)：添加异步操作的监听器
- 连接建立后
- 通道关闭后
注意：在完成连接建立后通道才能进行操作
/** * 客户端方式一：主线程同步等待连接建立，然后写入并刷出数据 */ public static void main(String[] args) throws InterruptedException { ChannelFuture channelFuture = new Bootstrap() .group(new NioEventLoopGroup()) .channel(NioSocketChannel.class) .handler(new ChannelInitializer<NioSocketChannel>() { @Override protected void initChannel(NioSocketChannel ch) throws Exception { ch.pipeline().addLast(new StringEncoder()); } }) .connect(new InetSocketAddress("127.0.0.1", 8080)); //1.channel() Channel beforeConnect = channelFuture.channel(); beforeConnect.writeAndFlush("连接建立前！"); System.out.println(beforeConnect); //[id: 0xf9941f21] //方式一：主线程同步等待 //2.sync() channelFuture.sync(); Channel afterConnect = channelFuture.channel(); afterConnect.writeAndFlush("连接建立后：主线程写出！"); System.out.println(afterConnect); //[id: 0xf9941f21, L:/127.0.0.1:61900 - R:/127.0.0.1:8080] }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
/** * 客户端方式二：主线程异步等待连接建立，使用另一线程写入并刷出数据 */ public static void main(String[] args) throws InterruptedException { ChannelFuture channelFuture = new Bootstrap() .group(new NioEventLoopGroup()) .channel(NioSocketChannel.class) .handler(new ChannelInitializer<NioSocketChannel>() { @Override protected void initChannel(NioSocketChannel ch) throws Exception { ch.pipeline().addLast(new StringEncoder()); } }) .connect(new InetSocketAddress("127.0.0.1", 8080)); //1.channel() Channel beforeConnect = channelFuture.channel(); beforeConnect.writeAndFlush("连接建立前！"); System.out.println(beforeConnect); //[id: 0xaf48a651] //方式二：主线程异步等待 //3.addListener(GenericFutureListener) channelFuture.addListener(new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture channelFuture) throws Exception { channelFuture.channel().writeAndFlush("连接建立后：另一线程写出"); System.out.println(channelFuture.channel()); //[id: 0x5201a68b, L:/127.0.0.1:62337 - R:/127.0.0.1:8080] } }); }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
3.2.2 Channel
Channel channel=channelFuture.channel();
- close() ：关闭 channel
- closeFuture() ：获取关闭通道
- pipeline() ：添加处理器
- write() ：写入数据
- writeAndFlush() ：写入并刷出数据
3.2.3 通道的关闭
同步善后
/** * 客户端-同步善后 */ public static void main(String[] args) throws InterruptedException { ChannelFuture channelFuture = new Bootstrap() .group(new NioEventLoopGroup()) .channel(NioSocketChannel.class) .handler(new ChannelInitializer<NioSocketChannel>() { @Override protected void initChannel(NioSocketChannel ch) throws Exception { ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG)); ch.pipeline().addLast(new StringEncoder()); } }) .connect(new InetSocketAddress("127.0.0.1", 8080)); Channel channel = channelFuture.sync().channel(); new Thread(()->{ Scanner scanner=new Scanner(System.in); while(true){ String s = scanner.nextLine(); if("q".equalsIgnoreCase(s)) break; channel.writeAndFlush(s); } //客户端输入任务完成，关闭通道 channel.close(); },"write").start(); //方式一：主线程同步等待通道关闭，进行善后处理 channel.closeFuture().sync(); log.debug("通道关闭，进行善后处理..."); }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
异步善后
/** * 客户端-异步善后 */ public static void main(String[] args) throws InterruptedException { NioEventLoopGroup group = new NioEventLoopGroup(); ChannelFuture channelFuture = new Bootstrap() .group(group) .channel(NioSocketChannel.class) .handler(new ChannelInitializer<NioSocketChannel>() { @Override protected void initChannel(NioSocketChannel ch) throws Exception { ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG)); ch.pipeline().addLast(new StringEncoder()); } }) .connect(new InetSocketAddress("127.0.0.1", 8080)); Channel channel = channelFuture.sync().channel(); new Thread(()->{ Scanner scanner=new Scanner(System.in); while(true){ String s = scanner.nextLine(); if("q".equalsIgnoreCase(s)) break; channel.writeAndFlush(s); } //客户端输入任务完成，关闭通道 channel.close(); },"write").start(); ChannelFuture closeFuture = channel.closeFuture(); closeFuture.addListener(new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture channelFuture) throws Exception { log.debug("通道关闭，进行善后处理..."); group.shutdownGracefully(); //优化关闭事件循环组 } }); }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
3.3 Future & Promise
3.3.1 概述
在异步处理时，经常用到这两个接口
Netty Promise —>(继承) Netty Future —>(继承) Jdk Future
- jdk Future：只能同步等待任务结束（或成功、或失败）才能得到结果
- netty Future：可以同步等待任务结束得到结果，也可以异步方式得到结果，但都是要等任务结束
- netty Promise：不仅有 netty Future 的功能，而且脱离了任务独立存在，只作为两个线程间传递结果的容器
功能/名称 jdk Future netty Future Promise cancel 取消任务 - - isCanceled 任务是否取消 - - isDone 任务是否完成，不能区分成功失败 - - get 获取任务结果，阻塞等待。如果任务失败，抛出异常 - - getNow - 获取任务结果，非阻塞，还未产生结果时返回 null - await - 等待任务结束，如果任务失败，不会抛异常，而是通过 isSuccess 判断 - sync - 等待任务结束，如果任务失败，抛出异常 - isSuccess - 判断任务是否成功 - cause - 获取失败信息，非阻塞，如果没有失败，返回null - addLinstener - 添加回调，异步接收结果 - setSuccess - - 设置成功结果 setFailure - - 设置失败结果 3.3.2 同步成功
public static void main(String[] args) throws ExecutionException, InterruptedException { //1.创建 EventLoop EventLoop eventLoop = new NioEventLoopGroup(2).next(); //2.获取 Promise DefaultPromise<Integer> promise=new DefaultPromise<>(eventLoop); //3.主线程创建随机线程执行任务，将任务结果设置到 promise容器 中 eventLoop.execute(()->{ log.debug("执行任务！"); try { Thread.sleep(1000); } catch (Exception e) { e.printStackTrace(); } promise.setSuccess(50); }); //4.主线程同步获取结果 log.debug("主线程非阻塞获取结果:{}",promise.getNow()); log.debug("主线程同步阻塞获取结果:{}",promise.get()); }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
10:12:49 [DEBUG] [main] o.e.j.f.PromiseSynSuccess : 主线程非阻塞获取结果:null 10:12:49 [DEBUG] [nioEventLoopGroup-2-1] o.e.j.f.PromiseSynSuccess : 执行任务！ 10:12:50 [DEBUG] [main] o.e.j.f.PromiseSynSuccess : 主线程同步阻塞获取结果:50- 1
- 2
- 3
3.3.3 异步成功
public static void main(String[] args) { //1.创建 EventLoop EventLoop eventLoop = new NioEventLoopGroup(2).next(); //2.获取 Promise DefaultPromise<Integer> promise=new DefaultPromise<>(eventLoop); //3.主线程创建随机线程执行任务，将任务结果设置到 promise容器 中 eventLoop.execute(()->{ log.debug("执行任务！"); try { Thread.sleep(1000); } catch (Exception e) { e.printStackTrace(); } promise.setSuccess(50); }); //4.主线程异获取结果 promise.addListener(new GenericFutureListener<Future<? super Integer>>() { @Override public void operationComplete(Future<? super Integer> future) throws Exception { log.debug("主线程异步获取结果{}",future.getNow()); } }); log.debug("主线程在异步获取结果时处理其他事情..."); }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
10:15:59 [DEBUG] [main] o.e.j.f.PromiseAsynSuccess : 主线程在异步获取结果时处理其他事情... 10:15:59 [DEBUG] [nioEventLoopGroup-2-1] o.e.j.f.PromiseAsynSuccess : 执行任务！ 10:16:00 [DEBUG] [nioEventLoopGroup-2-1] o.e.j.f.PromiseAsynSuccess : 主线程异步获取结果50- 1
- 2
- 3
3.3.4 同步失败
sync & get
/** * get() * sync()：也会出现异常，只是 get 会再用 ExecutionException 包一层 */ public static void main(String[] args) throws ExecutionException, InterruptedException { //1.创建 EventLoop EventLoop eventLoop = new NioEventLoopGroup(2).next(); //2.获取 Promise DefaultPromise<Integer> promise = new DefaultPromise<>(eventLoop); //3.主线程创建随机线程执行任务，将任务结果设置到 promise容器 中 eventLoop.execute(() -> { log.debug("执行任务！") try { Thread.sleep(1000); } catch (Exception e) { e.printStackTrace(); } promise.setFailure(new RuntimeException("同步失败")); }); //4.主线程同步获取结果 log.debug("主线程非阻塞获取结果:{}", promise.getNow()); log.debug("主线程同步阻塞获取结果:{}", promise.get()); }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
10:21:03 [DEBUG] [main] o.e.j.f.PromiseSynFail : 主线程非阻塞获取结果:null 10:21:03 [DEBUG] [nioEventLoopGroup-2-1] o.e.j.f.PromiseSynFail : 执行任务！ Exception in thread "main" java.util.concurrent.ExecutionException: java.lang.RuntimeException: 同步失败 at io.netty.util.concurrent.AbstractFuture.get(AbstractFuture.java:41) at org.example.java.future.PromiseSynFail.main(PromiseSynFail.java:30) Caused by: java.lang.RuntimeException: 同步失败 at org.example.java.future.PromiseSynFail.lambda$main$0(PromiseSynFail.java:25) at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:163) at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:416) at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:515) 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:750)- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
await
/** * await():不会抛异常 */ @Test public void test() throws InterruptedException { //1.创建 EventLoop EventLoop eventLoop = new NioEventLoopGroup(2).next(); //2.获取 Promise DefaultPromise<Integer> promise=new DefaultPromise<>(eventLoop); //3.主线程创建随机线程执行任务，将任务结果设置到 promise容器 中 eventLoop.execute(()->{ log.debug("执行任务！"); try { Thread.sleep(1000); } catch (Exception e) { e.printStackTrace(); } promise.setFailure(new RuntimeException("同步失败")); }); //4.主线程同步获取结果 log.debug("主线程非阻塞获取结果:{}",promise.getNow()); promise.await(); log.debug("主线程同步阻塞获取结果:{}",promise.isSuccess()?promise.getNow():promise.cause().toString()); }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
10:26:56 [DEBUG] [main] o.e.j.f.PromiseSynFail : 主线程非阻塞获取结果:null 10:26:56 [DEBUG] [nioEventLoopGroup-2-1] o.e.j.f.PromiseSynFail : 执行任务！ 10:26:57 [DEBUG] [main] o.e.j.f.PromiseSynFail : 主线程同步阻塞获取结果:java.lang.RuntimeException: 同步失败- 1
- 2
- 3
3.3.5 异步失败
/** * 不会抛异常 */ public static void main(String[] args) { //1.创建 EventLoop EventLoop eventLoop = new NioEventLoopGroup(2).next(); //2.获取 Promise DefaultPromise<Integer> promise=new DefaultPromise<>(eventLoop); //3.主线程创建随机线程执行任务，将任务结果设置到 promise容器 中 eventLoop.execute(()->{ log.debug("执行任务！"); try { Thread.sleep(1000); } catch (Exception e) { e.printStackTrace(); } promise.setFailure(new RuntimeException("异步失败")); }); //4.主线程异获取结果 promise.addListener(new GenericFutureListener<Future<? super Integer>>() { @Override public void operationComplete(Future<? super Integer> future) throws Exception { log.debug("主线程异步获取结果{}",promise.getNow()); log.debug("主线程异步获取结果{}",promise.isSuccess()?promise.getNow():promise.cause().toString()); } }); log.debug("主线程在异步获取结果时处理其他事情..."); }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
10:33:26 [DEBUG] [main] o.e.j.f.PromiseAsynFail : 主线程在异步获取结果时处理其他事情... 10:33:26 [DEBUG] [nioEventLoopGroup-2-1] o.e.j.f.PromiseAsynFail : 执行任务！ 10:33:27 [DEBUG] [nioEventLoopGroup-2-1] o.e.j.f.PromiseAsynFail : 主线程异步获取结果null 10:33:27 [DEBUG] [nioEventLoopGroup-2-1] o.e.j.f.PromiseAsynFail : 主线程异步获取结果java.lang.RuntimeException: 异步失败- 1
- 2
- 3
- 4
3.3.6 await 死锁检查
public static void main(String[] args) { //1.创建 EventLoop EventLoop eventLoop = new NioEventLoopGroup(2).next(); //2.获取 Promise DefaultPromise<Integer> promise=new DefaultPromise<>(eventLoop); //3.主线程创建随机线程执行任务，将任务结果设置到 promise容器 中 eventLoop.execute(()->{ System.out.println("1"); try { promise.await(); // 注意不能仅捕获 InterruptedException 异常 // 否则 死锁检查抛出的 BlockingOperationException 会继续向上传播 // 而提交的任务会被包装为 PromiseTask，它的 run 方法中会 catch 所有异常然后设置为 Promise 的失败结果而不会抛出 } catch (Exception e) { e.printStackTrace(); } System.out.println("2"); }); eventLoop.submit(()->{ System.out.println("3"); try { promise.await(); } catch (Exception e) { e.printStackTrace(); } System.out.println("4"); }); }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
1 2 3 4 io.netty.util.concurrent.BlockingOperationException: DefaultPromise@3c788ced(incomplete) at io.netty.util.concurrent.DefaultPromise.checkDeadLock(DefaultPromise.java:384) at io.netty.util.concurrent.DefaultPromise.await(DefaultPromise.java:212) at org.example.java.future.PromiseAwaitDeadlock.lambda$main$0(PromiseAwaitDeadlock.java:21) at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:163) at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:416) at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:515) 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:750) io.netty.util.concurrent.BlockingOperationException: DefaultPromise@3c788ced(incomplete) at io.netty.util.concurrent.DefaultPromise.checkDeadLock(DefaultPromise.java:384) at io.netty.util.concurrent.DefaultPromise.await(DefaultPromise.java:212) at org.example.java.future.PromiseAwaitDeadlock.lambda$main$1(PromiseAwaitDeadlock.java:33) at io.netty.util.concurrent.PromiseTask$RunnableAdapter.call(PromiseTask.java:38) at io.netty.util.concurrent.PromiseTask.run(PromiseTask.java:73) at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:163) at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:416) at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:515) 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:750)- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
3.4.2 Handler
- FixedLengthFrameDecoder：固定长度消息处理器
- LineBasedFrameDecoder：换行、回车分隔消息处理器
- DelimiterBasedFrameDecoder：自定义分隔符消息处理器
- LengthFieldBasedFrameDecoder：预设长度的消息处理器
- MessageCodeSharable：自定义协议编解码处理器，非jdk自带
- IdleStateHandler：生成假死事件消息处理器
- ChannelDuplexHandler：同时作为出站、入站的消息处理器
- SimpleChannelInboundHandler：指定消息类型的入站处理器
3.4.3 事件
- channelRead：读事件
- channelActive：连接建立事件
- channelInactive：连接断开事件
- exceptionCaught：出现异常事件
- userEventTriggered：触发特殊（自定义）事件
3.4 Handler & Pipeline
ChannelHandler 用来处理 Channel 上的各种事件，分为入站、出站两种。所有 ChannelHandler 被连成一串，就是 Pipeline
- 入站处理器：通常是 ChannelInboundHandlerAdapter 的子类，主要用来读取客户端数据，写回结果
- 出站处理器：通常是 ChannelOutboundHandlerAdapter 的子类，主要对写回结果进行加工
- 调试工具类：EmbeddedChannel
打个比喻，每个 Channel 是一个产品的加工车间，Pipeline 是车间中的流水线，ChannelHandler 就是流水线上的各道工序，而后面要讲的 ByteBuf 是原材料，经过很多工序的加工：先经过一道道入站工序，再经过一道道出站工序最终变成产品
Handler执行顺序
/** * 注意： * 1.写数据时应该使用channel写，才会从内部往外部的handler传递处理 * 2.处理到通道的数据应该是ByteBuf */ new ServerBootstrap() .group(new NioEventLoopGroup(),new NioEventLoopGroup(2)) .channel(NioServerSocketChannel.class) .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { ch.pipeline().addLast("h1",new ChannelInboundHandlerAdapter(){ @Override public void channelRead(ChannelHandlerContext ctx, Object msg) { System.out.println(1); String name=((ByteBuf)msg).toString(Charset.forName("UTF-8")); //1.启动下一个入站handler，并传递处理结果 ctx.fireChannelRead(name); } }); ch.pipeline().addLast("h2",new ChannelInboundHandlerAdapter(){ @Override public void channelRead(ChannelHandlerContext ctx, Object name) { System.out.println(2); //2.启动下一个入站handler，并传递处理结果 ctx.fireChannelRead(name); } }); ch.pipeline().addLast("h3",new ChannelInboundHandlerAdapter(){ @Override public void channelRead(ChannelHandlerContext ctx, Object name) { System.out.println(3); //3.从内部开始启动第一个出站handler，并传递处理结果 ch.writeAndFlush(ctx.alloc().buffer().writeBytes(((String)name).getBytes())); //ctx.channel().writeAndFlush(ctx.alloc().buffer().writeBytes(((String)name).getBytes())); //注意：该操作是从当前位置往外找处理器，而外面没有出站处理器导致内部的出站处理器无法执行 //ctx.write(ctx.alloc().buffer().writeBytes(((String)name).getBytes()), promise); } }); ch.pipeline().addLast("h4",new ChannelOutboundHandlerAdapter(){ @Override public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) { System.out.println(4); //4.往外启动下一个出站handler，并传递处理结果 ctx.write(msg, promise); } }); ch.pipeline().addLast("h5",new ChannelOutboundHandlerAdapter(){ @Override public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) { System.out.println(5); //5.往外启动下一个出站handler，并传递处理结果 ctx.write(msg, promise); } }); ch.pipeline().addLast("h6",new ChannelOutboundHandlerAdapter(){ @Override public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) { System.out.println(6); //6.往外启动下一个出站handler，并传递处理结果 ctx.write(msg, promise); } }); } }) .bind(8080);- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
1 2 3 6 5 4- 1
- 2
- 3
- 4
- 5
- 6
3.5 ByteBuf
💡 优势
- 可以自动扩容
- 池化：可以重用池中 ByteBuf 实例，更节约内存，减少内存溢出的可能
- 读写指针分离：不需要像 ByteBuffer 一样切换读写模式
- 支持链式调用：使用更流畅
- 很多地方体现零拷贝：例如 slice、duplicate、CompositeByteBuf
1）创建
/** * - 不指定容量时，默认为256 * - 创建了一个默认的 ByteBuf（池化基于直接内存的 ByteBuf），初始容量是 10 */ ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(10); System.out.println(buffer);- 1
- 2
- 3
- 4
- 5
- 6
PooledUnsafeDirectByteBuf(ridx: 0, widx: 0, cap: 10)- 1
调试方法
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()); }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
2）扩容规则
- 如何写入后数据大小未超过 512，则选择下一个 16 的整数倍，例如写入后大小为 12 ，则扩容后 capacity 是 16
- 如果写入后数据大小超过 512，则选择下一个 2^n，例如写入后大小为 513，则扩容后 capacity 是 210=1024（29=512 已经不够了）
- 扩容不能超过 max capacity 会报错
3）直接内存 vs 堆内存
- 直接内存：创建和销毁的代价昂贵，但 读写性能高（少一次内存复制），适合配合池化功能一起用
- 直接内存：对 GC 压力小，因为这部分内存不受 JVM 垃圾回收的管理，但也要注意及时主动释放
//堆内存 ByteBuf buffer = ByteBufAllocator.DEFAULT.heapBuffer(10); //直接内存 ByteBuf buffer = ByteBufAllocator.DEFAULT.directBuffer(10);- 1
- 2
- 3
- 4
4）池化 vs 非池化
池化的最大意义：重用 ByteBuf
池化功能是否开启，可以通过系统环境变量来设置：-Dio.netty.allocator.type={unpooled|pooled}
创建实例 高并发时 分配效率 直接内存 堆内存 有了池化 重用池中的实例 节约内存，减少内存溢出 采用了类似 jemalloc 的内存分配算法提升分配效率 没有池化 创建新的实例 创建、销毁的代价昂贵 增加 GC 压力 - 4.1 以后：非 Android 平台默认启用池化实现，Android 平台启用非池化实现
- 4.1 之前：池化功能还不成熟，默认是非池化实现
5）组成
- 为什么设置可扩容部分？
- 按需分配，合理利用内存
- 组成部分：
- 最开始读写指针都在 0 位置
6）写入方法
- 这些方法中未指明返回值的，其返回值都是 ByteBuf，意味着可以链式调用
- 网络传输，默认习惯是 Big Endian
- 还有一类方法是 set 开头 的一系列方法，也可以写入数据，但不会改变写指针位置
方法签名 含义 备注 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) 写入字符串 写入字符串 markReaderIndex() 标记读指针 resetReaderIndex() 重置读指针 使用案例
//1.先写入 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 |.... | +--------+-------------------------------------------------+----------------+ //2.再写入一个 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 |........ | +--------+-------------------------------------------------+----------------+ //3.再写入一个 int 整数时，容量不够了（初始容量是 10），这时会引发扩容 buffer.writeInt(6); log(buffer); //结果是： 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 |............ | +--------+-------------------------------------------------+----------------+- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
7）读取方法
使用案例
- 注意：案例中 log() 不会打印废弃区域
//1.读取 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 |........ | +--------+-------------------------------------------------+----------------+ //2.如果需要重复读取 int 整数 5，怎么办？ //方式一：可以在 read 前先做个标记 mark //方式二：采用 get 开头的一系列方法，不会改变 read index 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 |.... | +--------+-------------------------------------------------+----------------+ //3.这时要重复读取的话，重置到标记位置 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 |........ | +--------+-------------------------------------------------+----------------+- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
8）retain & release
由于 Netty 中有堆外内存的 ByteBuf 实现，堆外内存最好是手动来释放，而不是等 GC 垃圾回收。
- UnpooledHeapByteBuf：使用的是 JVM 内存，只需等 GC 回收内存即可
- UnpooledDirectByteBuf：使用的就是直接内存了，需要特殊的方法回收内存
- PooledByteBuf：使用了池化机制，需要更复杂的规则回收内存
回收内存的源码实现，请关注此方法的不同实现：
protected abstract void deallocate()引用计数法
- 每个 ByteBuf 都实现了 ReferenceCounted 接口
- 每个 ByteBuf 对象的初始计数为 1
- release()：计数减 1，如果计数为 0，ByteBuf 内存被回收
- retain()：计数加 1，表示调用者没用完之前，其它 handler 即使调用了 release 也不会造成回收
当计数为 0 时，底层内存会被回收，这时即使 ByteBuf 对象还在，其各个方法均无法正常使用
释放规则
在 pipeline 中，ByteBuf 一般需要传递给下一个 handler 。所以释放规则是，谁是最后使用者，谁负责 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）
//io.netty.channel.DefaultChannelPipeline.TailContext ——> public void channelRead(ChannelHandlerContext ctx, Object msg) { DefaultChannelPipeline.this.onUnhandledInboundMessage(ctx, msg); } protected void onUnhandledInboundMessage(ChannelHandlerContext ctx, Object msg) { this.onUnhandledInboundMessage(msg); if (logger.isDebugEnabled()) { logger.debug("Discarded message pipeline : {}. Channel : {}.", ctx.pipeline().names(), ctx.channel()); } } 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); } } public static boolean release(Object msg) { return msg instanceof ReferenceCounted ? ((ReferenceCounted)msg).release() : false; }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
-
-
出站 ByteBuf 处理原则
- 出站消息最终都会转为 ByteBuf 输出，一直向前传，由 HeadContext flush 后 release
-
**异常 **处理原则
- 有时候不清楚 ByteBuf 被引用了多少次，但又必须彻底释放，可以循环调用 release 直到返回 true
9）零拷贝-slice
对原始 ByteBuf 切片成多个 ByteBuf，切片后的 ByteBuf 并没有发生内存复制，还是使用原始 ByteBuf 的内存，切片后的 ByteBuf 维护独立的 read，write 指针
- slice()：read index - write index
- slice(int,int)：int - int
注意：切片后的 max capacity 被固定为这个区间的大小，因此不能追加 write
使用案例
//1.初始化 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 |... | +--------+-------------------------------------------------+----------------+ //2.切片：如果写入会报 IndexOutOfBoundsException 异常 ByteBuf slice = origin.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 |... | +--------+-------------------------------------------------+----------------+ //3.原始 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 |.. | +--------+-------------------------------------------------+----------------+ //4.切片得到的 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 |... | +--------+-------------------------------------------------+----------------+ //5.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 |... | +--------+-------------------------------------------------+----------------+ //6.原始 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 |.. | +--------+-------------------------------------------------+----------------+- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
10）零拷贝-duplicate
- duplicate()：0 - capacity
注意：没有 max capacity 的限制，只是前部分与原始 ByteBuf 使用同一块底层内存，且读写指针独立
11）深拷贝-copy
- copy()：会将底层内存数据进行深拷贝，因此无论读写，都与原始 ByteBuf 无关
12）零拷贝-CompositeByteBuf
CompositeByteBuf 是一个组合的 ByteBuf，它内部维护了一个 Component 数组，每个 Component 管理一个 ByteBuf，记录了这个 ByteBuf 相对于整体偏移量等信息，代表着整体中某一段的数据。
- addComponents(boolean b,ByteBuf… buffers)：
- 可以将多个 ByteBuf 合并为一个逻辑上的 ByteBuf，避免拷贝
- 自动递增 write index, 默认 write index 始终为 0
- 优点：对外是一个虚拟视图，组合这些 ByteBuf 不会产生内存复制
- 缺点：复杂了很多，多次操作会带来性能的损耗
使用案例
//1.有两个 ByteBuf 如下 ByteBuf buf1 = ByteBufAllocator.DEFAULT.buffer(5); ByteBuf buf2 = ByteBufAllocator.DEFAULT.buffer(5); buf1.writeBytes(new byte[]{1, 2, 3, 4, 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 |..... | +--------+-------------------------------------------------+----------------+ //2.现在需要一个新的 ByteBuf，内容来自于刚才的 buf1 和 buf2，如何实现？ //方式一：深拷贝 //方式二：零拷贝 CompositeByteBuf buf3 = ByteBufAllocator.DEFAULT.compositeBuffer(); buf3.addComponents(true, buf1, buf2); //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 |.......... | +--------+-------------------------------------------------+----------------+- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
13）零拷贝-Unpooled
Unpooled 是一个工具类，类如其名，提供了非池化的 ByteBuf 创建、组合、复制等操作
- wrappedBuffer (ByteBuf… buffers)：包装 ByteBuf
- wrappedBuffer (byte[]… arrays)：包装 byte[]
注意：当包装个数超过一个时, 底层使用了 CompositeByteBuf
使用案例
//1.有两个 ByteBuf 如下 ByteBuf buf1 = ByteBufAllocator.DEFAULT.buffer(5); ByteBuf buf2 = ByteBufAllocator.DEFAULT.buffer(5); buf1.writeBytes(new byte[]{1, 2, 3, 4, 5}); buf2.writeBytes(new byte[]{6, 7, 8, 9, 10}); //2.包装 ByteBuf 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 |.......... | +--------+-------------------------------------------------+----------------+ //3.包装普通字节数组 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 |...... | +--------+-------------------------------------------------+----------------+- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
四、双向通信
4.1 Netty实现
实现一个 echo server
服务器
//1.创建 非IO事件循环组、IO事件循环组、接收事件循环组 DefaultEventLoopGroup defaultGroup=new DefaultEventLoopGroup(2); NioEventLoopGroup group=new NioEventLoopGroup(2); //2.监听 8080 端口 ChannelFuture channelFuture = new ServerBootstrap() .channel(NioServerSocketChannel.class) .group(new NioEventLoopGroup(),group) .childHandler(new ChannelInitializer<NioSocketChannel>() { @Override protected void initChannel(NioSocketChannel ch) throws Exception { ch.pipeline().addLast(defaultGroup,new StringDecoder(Charset.forName("UTF-8"))); //接收信息 ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){ @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { System.out.println("客户端："+msg); //将信息进行封装 ByteBuf buf=(ByteBuf) msg; ByteBuf message = ctx.alloc().buffer(); message.writeBytes("服务器：".getBytes()); message.writeBytes(buf); //思考：释放 msg 缓冲区。无需释放 message ，由 HeadContext 进行释放。 buf.release(); ctx.channel().writeAndFlush(message); } }); //发出信息 ch.pipeline().addLast(new ChannelOutboundHandlerAdapter(){ @Override public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { //无需释放 msg ，由 HeadContext 进行释放 ctx.write(msg); } }); ch.pipeline().addLast(defaultGroup,new StringEncoder(Charset.forName("UTF-8"))); } }) .bind(8080); channelFuture.addListener(new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture channelFuture) throws Exception { System.out.println("服务器启动成功！"); } });- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
客户端
//1.创建 非IO事件循环组、IO事件循环组 DefaultEventLoopGroup defaultGroup = new DefaultEventLoopGroup(2); NioEventLoopGroup group = new NioEventLoopGroup(2); //2.连接 127.0.0.1：8080 ChannelFuture connect = new Bootstrap() .channel(NioSocketChannel.class) .group(group) .handler(new ChannelInitializer<NioSocketChannel>() { @Override protected void initChannel(NioSocketChannel ch) throws Exception { ch.pipeline().addLast(new StringDecoder(Charset.forName("UTF-8"))); ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){ @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { System.out.println(msg); System.out.print("客户端："); //思考：无需释放 msg ，由 TailContext 进行释放。 ctx.fireChannelRead(msg); } }); ch.pipeline().addLast(new StringEncoder(Charset.forName("UTF-8"))); } }) .connect(new InetSocketAddress("127.0.0.1", 8080)); //3.等待连接建立后执行操作 Channel channel = connect.sync().channel(); //4.接收控制台输入，并发送 defaultGroup.submit(()->{ System.out.println("客户端启动完成！"); Scanner in=new Scanner(System.in); String message=""; System.out.print("客户端："); while(!"q".equalsIgnoreCase(message)){ message=in.nextLine(); channel.writeAndFlush(message); } //5.关闭连接 channel.close(); }); //6.关闭连接之后释放资源 ChannelFuture closeFuture = channel.closeFuture(); closeFuture.addListener(new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture channelFuture) throws Exception { defaultGroup.shutdownGracefully(); group.shutdownGracefully(); } });- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
4.2 Bio实现💡
我最初在认识上有这样的误区，认为只有在 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(); } }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
客户端
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(); } }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
-
相关阅读:
蓝牙 - 苹果iOS所支持的profile
JS构造函数和原型
C++ 图解二叉树非递归后序 + 实战力扣题
IMX6ULL学习笔记（7）——通过SD卡启动U-Boot
树的相关知识的注意事项
多维数据库中的高效计算机制是什么？
想要顺利拿下订单，外贸人的价格谈判思维你GET了吗？
互联网资讯查询易语言代码
服装商城网站 毕业设计-附源码241505
【Learning eBPF-3】一个 eBPF 程序的深入剖析
- 原文地址：https://blog.csdn.net/weixin_43401592/article/details/127586357
