• Netty：入门（2）

    相关文章：

    《IO 模型与多路复用》

    《Java NIO》

    《Netty：入门（1）》

            写在开头：本文为学习后的总结，可能有不到位的地方，错误的地方，欢迎各位指正。

    前言

            在前文中，我们对Netty的内容做了简单的介绍，本文我们会结合Netty的流程图相对深入一些的介绍下其中的重要组件。

    目录

    前言

    一、EventLoop

            EventLoop

            EventLoopGroup

            EventLoopGroup分工  

            分工细化

             EventLoopGroup切换源码

     二、 Channel

            Channel 的用处

            如何建立连接

            sync

            addListener

            如何关闭连接

            sync

            addListener

     三、Future与Promise

            Future的介绍

            JDK Future

            Netty Future

            Netty Promise

     四、Handler与Pipeline

            Pipeline

            OutboundHandler

            socketChannel.writeAndFlush()

            ctx.writeAndFlush()

     五、ByteBuf

            创建     

            直接内存与堆内存

             池化与非池化

            ByteBuf组成

            最大容量与当前容量

            读指针和写指针

             读写

            扩容

             内存释放

            释放方法

            释放规则

            切片

    一、EventLoop

            EventLoop

            事件循环对象 EventLoop本质是一个单线程执行器（同时维护了一个 Selector），支持异步提交执行任务，线程启动时会调用 NioEventLoop 的 run 方法，执行 I/O 任务和非 I/O 任务：

    • I/O 任务，即 selectionKey 中 ready 的事件，如 accept、connect、read、write 等，由 processSelectedKeys 方法触发。
    • 非 IO 任务，添加到 taskQueue 中的任务，如 register0、bind0 等任务，由 runAllTasks 方法触发。

            它的继承关系如下：

    • 继承自 j.u.c.ScheduledExecutorService，因此包含了线程池中所有的方法
    • 继承自 netty 自己的 OrderedEventExecutor，提供了 boolean inEventLoop(Thread thread) 方法判断一个线程是否属于此 EventLoop，同时还提供了 EventLoopGroup parent() 方法来看看自己属于哪个 EventLoopGroup

            EventLoopGroup

            事件循环组EventLoopGroup是一组 EventLoop，Channel 一般会调用 EventLoopGroup 的 register 方法来绑定其中一个 EventLoop，后续这个 Channel 上的 io 事件都由此 EventLoop 来处理（保证了 io 事件处理时的线程安全）。

            继承自 netty 自己的 EventExecutorGroup，实现了 Iterable 接口提供遍历 EventLoop 的，并提供了next方法获取集合中下一个 EventLoop。

            EventLoop初始化时可以指定线程数，可以不指定，如果不指定，这里会传0用来调用父类MultithreadEventLoopGroup中得构造方法。

    1. public class NioEventLoopGroup extends MultithreadEventLoopGroup {
    2.     public NioEventLoopGroup() {
    3.         this(0);
    4.     }
    5.     public NioEventLoopGroup(int nThreads) {
    6.         this(nThreads, (Executor) null);
    7.     }
    8.     ...
    9. }

            MultithreadEventLoopGroup类提供1个默认线程数DEFAULT_EVENT_LOOP_THREADS，比较1与Netty系统参数"io.netty.eventLoopThreads"（如未设置则使用CPU核心数*2）中得最大值。当传入得初始化线程数为0时就使用这个默认线程数。

    1. public abstract class MultithreadEventLoopGroup extends MultithreadEventExecutorGroup implements EventLoopGroup {
    2.  
    3.     private static final InternalLogger logger = InternalLoggerFactory.getInstance(MultithreadEventLoopGroup.class);
    4.  
    5.     // 默认线程数
    6.     private static final int DEFAULT_EVENT_LOOP_THREADS;
    7.  
    8.     static {
    9.         // 静态代码块中，给默认线程数DEFAULT_EVENT_LOOP_THREADS 赋值
    10.         // 比较1与Netty系统参数"io.netty.eventLoopThreads"（如未设置则使用CPU核心数*2）中得最大值
    11.         DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt(
    12.                 "io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));
    13.  
    14.         if (logger.isDebugEnabled()) {
    15.             logger.debug("-Dio.netty.eventLoopThreads: {}", DEFAULT_EVENT_LOOP_THREADS);
    16.         }
    17.     }
    18.  
    19.     // 如果传入得线程数为0，则使用默认线程数
    20.     protected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... args) {
    21.         super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);
    22.     }
    23.     ...
    24. }

            EventLoop处理普通与定时任务

    1. public class TestEventLoop {
    2.     public static void main(String[] args) {
    3.         // 创建拥有两个EventLoop的NioEventLoopGroup，对应两个线程
    4.         EventLoopGroup group = new NioEventLoopGroup(2);
    5.         // 通过next方法可以获得下一个 EventLoop
    6.         System.out.println(group.next());
    7.         System.out.println(group.next());
    8.  
    9.         // 通过EventLoop执行普通任务
    10.         group.next().execute(()->{
    11.             System.out.println(Thread.currentThread().getName() + " hello");
    12.         });
    13.  
    14.         // 通过EventLoop执行定时任务（表示立即执行（0），每次间隔1秒（1））
    15.         group.next().scheduleAtFixedRate(()->{
    16.             System.out.println(Thread.currentThread().getName() + " hello2");
    17.         }, 0, 1, TimeUnit.SECONDS);
    18.         
    19.         // 优雅地关闭
    20.         group.shutdownGracefully();
    21.     }
    22. }

            输出结果如下

    1. io.netty.channel.nio.NioEventLoop@7bb11784
    2. io.netty.channel.nio.NioEventLoop@33a10788
    3. nioEventLoopGroup-2-1 hello
    4. nioEventLoopGroup-2-2 hello2
    5. nioEventLoopGroup-2-2 hello2
    6. nioEventLoopGroup-2-2 hello2

            优雅关闭 shutdownGracefully 方法。该方法会首先切换 EventLoopGroup 到关闭状态从而拒绝新的任务的加入，然后在任务队列的任务都处理完成后，停止线程的运行。从而确保整体应用是在正常有序的状态下退出的。

            下面演示下处理IO任务：

            服务器代码

    1. public class MyServer {
    2.     public static void main(String[] args) {
    3.         new ServerBootstrap()
    4.                 // 添加EventLoop
    5.                 .group(new NioEventLoopGroup())
    6.                 // 选择服务端Channel实现
    7.                 .channel(NioServerSocketChannel.class)
    8.                 // 添加处理器
    9.                 .childHandler(new ChannelInitializer() {
    10.                     @Override    // 连接建立后调用
    11.                     protected void initChannel(SocketChannel socketChannel) throws Exception {
    12.                         socketChannel.pipeline().addLast(new ChannelInboundHandlerAdapter() {
    13.                             @Override
    14.                             public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    15.                                 // 将服务端收到的数据转bytebuf后再转String
    16.                                 ByteBuf buf = (ByteBuf) msg;
    17.                                 log.debug(buf.toString(Charset.defaultCharset()));
    18.  
    19.                             }
    20.                         });
    21.                     }
    22.                 })
    23.                 .bind(8080);
    24.     }
    25. }

             客户端代码

    1. public class MyClient {
    2.     public static void main(String[] args) throws IOException, InterruptedException {
    3.         Channel channel = new Bootstrap()
    4.                 // 添加EventLoop
    5.                 .group(new NioEventLoopGroup())
    6.                 // 选择客户端Channel实现
    7.                 .channel(NioSocketChannel.class)
    8.                 // 添加处理器
    9.                 .handler(new ChannelInitializer() {
    10.                     @Override    // 连接建立后调用
    11.                     protected void initChannel(SocketChannel socketChannel) throws Exception {
    12.                         socketChannel.pipeline().addLast(new StringEncoder());
    13.                     }
    14.                 })
    15.                 // 连接到服务器
    16.                 .connect(new InetSocketAddress("localhost", 8080))
    17.                 .sync()
    18.                 .channel();
    19.         System.out.println(channel);
    20.         System.in.read();
    21.     }
    22. }

            EventLoopGroup分工  

            Bootstrap的group()方法可以传入两个EventLoopGroup参数，分别负责处理Accept事件（boss线程）与Read/Write（worker线程）（参考下我们前一篇文章中介绍得流程图）

    1. public class MyServer {
    2.     public static void main(String[] args) {
    3.         new ServerBootstrap()
    4.             	// 两个Group，分别为Boss（负责Accept事件），Worker（负责读写事件）
    5.                 .group(new NioEventLoopGroup(1), new NioEventLoopGroup(2))
    6.                 // ...
    7. 				// 上文中处理IO任务得代码
    8.     }
    9. }

             使用上文中多个客户端分别发送 hello 结果

    1. nioEventLoopGroup-3-1 hello1
    2. nioEventLoopGroup-3-2 hello2
    3. nioEventLoopGroup-3-1 hello3
    4. nioEventLoopGroup-3-2 hello4
    5. nioEventLoopGroup-3-2 hello4

             可以看出，一个EventLoop可以负责多个Channel，且EventLoop一旦与Channel绑定，则一直负责处理该Channel中的事件。

            分工细化

            当有的任务需要较长的时间处理时，可以再添加一个非NioEventLoopGroup，避免同一个NioEventLoop中的其他Channel在较长的时间内都无法得到处理。

    1. public class MyServer {
    2.     public static void main(String[] args) {
    3.         // 增加自定义的非NioEventLoopGroup
    4.         EventLoopGroup group = new DefaultEventLoopGroup();
    5.         
    6.         new ServerBootstrap()
    7.                 .group(new NioEventLoopGroup(1), new NioEventLoopGroup(2))
    8.                 .channel(NioServerSocketChannel.class)
    9.                 .childHandler(new ChannelInitializer() {
    10.                     @Override
    11.                     protected void initChannel(SocketChannel socketChannel) throws Exception {
    12.                         // 增加两个handler，第一个使用NioEventLoopGroup处理，第二个使用自定义EventLoopGroup处理
    13.                         socketChannel.pipeline().addLast("nioHandler",new ChannelInboundHandlerAdapter() {
    14.                             @Override
    15.                             public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    16.                                 ByteBuf buf = (ByteBuf) msg;
    17.                                 System.out.println(Thread.currentThread().getName() + " " + buf.toString(StandardCharsets.UTF_8));
    18.                                 // 调用下一个handler
    19.                                 ctx.fireChannelRead(msg);
    20.                             }
    21.                         })
    22.                         // 该handler绑定自定义的Group
    23.                         .addLast(group, "myHandler", new ChannelInboundHandlerAdapter() {
    24.                             @Override
    25.                             public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    26.                                 ByteBuf buf = (ByteBuf) msg;
    27.                                 System.out.println(Thread.currentThread().getName() + " " + buf.toString(StandardCharsets.UTF_8));
    28.                             }
    29.                         });
    30.                     }
    31.                 })
    32.                 .bind(8080);
    33.     }
    34. }

             启动四个客户端发送数据

    1. nioEventLoopGroup-4-1 hello1
    2. defaultEventLoopGroup-2-1 hello1
    3. nioEventLoopGroup-4-2 hello2
    4. defaultEventLoopGroup-2-2 hello2
    5. nioEventLoopGroup-4-1 hello3
    6. defaultEventLoopGroup-2-3 hello3
    7. nioEventLoopGroup-4-2 hello4
    8. defaultEventLoopGroup-2-4 hello4

             可以看出，客户端与服务器之间的事件，被nioEventLoopGroup和defaultEventLoopGroup分别处理。

             EventLoopGroup切换源码

            不同的EventLoopGroup切换的实现原理如下：由上面的图可以看出，当handler中绑定的Group不同时，需要切换Group来执行不同的任务。

    1. static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {
    2.     final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next);
    3.     // 获取下一个handler得EventLoop 
    4.     EventExecutor executor = next.executor();
    5.     
    6.     // 判断当前handler中得线程与executor（下一个handler得EventLoop）是否为同一个
    7.     if (executor.inEventLoop()) {
    8.         // 使用当前EventLoopGroup中的EventLoop来处理任务
    9.         next.invokeChannelRead(m);
    10.     } else {
    11.         // 将要执行得代码作为任务交给executor处理
    12.         executor.execute(new Runnable() {
    13.             public void run() {
    14.                 next.invokeChannelRead(m);
    15.             }
    16.         });
    17.     }
    18. }
    • 如果两个 handler 绑定的是同一个EventLoopGroup，那么就直接调用
    • 否则，把要调用的代码封装为一个任务对象，由下一个 handler 的 EventLoopGroup 来调用

     二、 Channel

            Channel 的用处

            Channel在Netty中与NIO中一样，是数据的通道，Channel 的常用方法包括：

    • close()：用来关闭Channel
    • closeFuture()：用来处理 Channel 的关闭，关闭的方法根据同步与异步分别有sync与addListener实现。
    • pipeline()：用于添加处理器
    • write()：方法将数据写入（因为缓冲机制，数据被写入到 Channel 中以后，不会立即被发送。只有当缓冲满了或者调用了flush()方法后，才会将数据通过 Channel 发送出去）
    • writeAndFlush()：将数据写入并立即发送（刷出）

            如何建立连接

            在客户端代码中，我们使用connect来连接服务端，从而返回一个供我们用作数据传输的channel，建立连接(connect)的过程是异步非阻塞的，主线程调用后不会被阻塞，真正去执行连接操作的是NIO线程（即NioEventLoop 中的线程）。

    1. public class MyClient {
    2.     public static void main(String[] args) throws IOException, InterruptedException {
    3.         ChannelFuture channelFuture = new Bootstrap()
    4.                 .group(new NioEventLoopGroup())
    5.                 .channel(NioSocketChannel.class)
    6.                 .handler(new ChannelInitializer() {
    7.                     @Override
    8.                     protected void initChannel(SocketChannel socketChannel) throws Exception {
    9.                         socketChannel.pipeline().addLast(new StringEncoder());
    10.                     }
    11.                 })
    12.                 // 该方法为异步非阻塞方法，主线程调用后不会被阻塞，真正去执行连接操作的是NIO线程
    13.             	// NIO线程：NioEventLoop 中的线程
    14.                 .connect(new InetSocketAddress("localhost", 8080));
    15.         
    16.         
    17.         // 获取客户端-服务器之间的Channel对象
    18.         Channel channel = channelFuture.channel();
    19.         channel.writeAndFlush("hello world");
    20.         System.in.read();
    21.     }
    22. }

            这样将会导致一个问题，即代码虽然继续向下运行到了channelFuture.channel()，但此时我们并没有获取到channel，所以下文中的hello world是无法发送的。

            为了解决这个问题，Netty分别提供了同步与异步的解决方案。

            sync

            在connect后，使用channelFuture.sync()方法，阻塞主线程，同步处理结果，等待连接真正建立好以后，再去获得 Channel 传递数据。使用该方法，获取 Channel 和发送数据的线程都是主线程。

    1. // 该方法用于等待连接真正建立
    2. channelFuture.sync();
    3.         
    4. // 获取客户端-服务器之间的Channel对象
    5. Channel channel = channelFuture.channel();

            addListener

            异步获取建立连接后的 Channel 和发送数据，使得执行这些操作的线程是 NIO 线程。

    1. // 当connect方法执行完毕后，也就是连接真正建立后
    2. // 会在NIO线程中调用operationComplete方法
    3. channelFuture.addListener(new ChannelFutureListener() {
    4.     @Override
    5.     public void operationComplete(ChannelFuture channelFuture) throws Exception {
    6.         Channel channel = channelFuture.channel();
    7.         channel.writeAndFlush("hello world");
    8.         }
    9. });

            如何关闭连接

            当我们要关闭channel时，可以调用channel.close()方法进行关闭。但是该方法也是一个异步方法。真正的关闭操作并不是在调用该方法的线程中执行的，而是在NIO线程中执行真正的关闭操作。

            同连接的建立一样，如果我们想在channel真正关闭以后，执行一些额外的操作，可以选择同于与异步两种方法实现。

            sync

            通过channel.closeFuture()方法获得对应的ChannelFuture对象，然后调用sync()方法阻塞执行操作的线程，等待channel真正关闭后，再执行其他操作。

    1. // 获得closeFuture对象
    2. ChannelFuture closeFuture = channel.closeFuture();
    3.  
    4. // 同步等待NIO线程执行完close操作
    5. closeFuture.sync();

            addListener

            调用closeFuture.addListener方法，添加close的后续操作

    1. closeFuture.addListener(new ChannelFutureListener() {
    2.     @Override
    3.     public void operationComplete(ChannelFuture channelFuture) throws Exception {
    4.         // 等待channel关闭后才执行的操作
    5.         System.out.println("关闭之后执行一些额外操作...");
    6.         // 关闭EventLoopGroup
    7.         group.shutdownGracefully();
    8.     }
    9. });

            下面展示输入q后连接关闭的样例：

    1. public class ReadClient {
    2.     public static void main(String[] args) throws InterruptedException {
    3.         // 创建EventLoopGroup，使用完毕后关闭
    4.         NioEventLoopGroup group = new NioEventLoopGroup();
    5.         
    6.         ChannelFuture channelFuture = new Bootstrap()
    7.                 .group(group)
    8.                 .channel(NioSocketChannel.class)
    9.                 .handler(new ChannelInitializer() {
    10.                     @Override
    11.                     protected void initChannel(SocketChannel socketChannel) throws Exception {
    12.                         socketChannel.pipeline().addLast(new StringEncoder());
    13.                     }
    14.                 })
    15.                 .connect(new InetSocketAddress("localhost", 8080));
    16.         channelFuture.sync();
    17.  
    18.         Channel channel = channelFuture.channel();
    19.         Scanner scanner = new Scanner(System.in);
    20.  
    21.         // 创建一个线程用于输入并向服务器发送
    22.         new Thread(()->{
    23.             while (true) {
    24.                 String msg = scanner.next();
    25.                 if ("q".equals(msg)) {
    26.                     // 关闭操作是异步的，在NIO线程中执行
    27.                     channel.close();
    28.                     break;
    29.                 }
    30.                 channel.writeAndFlush(msg);
    31.             }
    32.         }, "inputThread").start();
    33.  
    34.         // 获得closeFuture对象
    35.         ChannelFuture closeFuture = channel.closeFuture();
    36.         System.out.println("waiting close...");
    37.         
    38.         // 同步等待NIO线程执行完close操作
    39.         closeFuture.sync();
    40.         
    41.         // 关闭之后执行一些操作，可以保证执行的操作一定是在channel关闭以后执行的
    42.         System.out.println("关闭之后执行一些额外操作...");
    43.         
    44.         // 关闭EventLoopGroup
    45.         group.shutdownGracefully();
    46.     }
    47. }

     三、Future与Promise

            Future的介绍

            JDK创建了Future来提供主线程与异步线程之间的交互功能，netty 中的Future与 jdk 中的 Future 一样，但又有所不同。

    • JDK Future 只能同步等待任务结束（或成功、或失败）才能得到结果
    • Netty Future 可以同步等待任务结束得到结果，也可以异步方式得到结果，但都是要等任务结束
    • Netty Promise 不仅有 Netty Future 的功能，而且脱离了任务独立存在，只作为两个线程间传递结果的容器

            JDK Future

            通过向线程池中提交任务，可以获得一个线程池的返回的future结果，主线程通过future来获取异步线程的执行结果。缺点在于，JDK中的future只能同步的获取结果，即必须要等到异步线程执行完。

    1. public class JdkFuture {
    2.     public static void main(String[] args) throws ExecutionException, InterruptedException {
    3.         // 创建线程池
    4.         ExecutorService service = Executors.newFixedThreadPool(2);
    5.  
    6.         // 提交任务，获得Future对象
    7.         Future future = service.submit(new Callable() {
    8.             @Override
    9.             public Integer call() throws Exception {
    10.                 log.debug("开始计算");
    11. 				Thread.sleep(1000);
    12.                 return 50;
    13.             }
    14.         });
    15.         // 主线程通过future来获取结果
    16. 		log.debug("等待结果");
    17. 		log.debug("结果是{}",future.get());
    18.     }
    19. }

            Netty Future

            在Netty中，Future可以异步的获取结果。

    1. public class NettyFuture {
    2.     public static void main(String[] args) throws ExecutionException, InterruptedException {
    3.         NioEventLoopGroup group = new NioEventLoopGroup();
    4.  
    5.         // 获得 EventLoop 对象
    6.         EventLoop eventLoop = group.next();
    7.         Future future = eventLoop.submit(new Callable() {
    8.             @Override
    9.             public Integer call() throws Exception {
    10. 				log.debug("开始计算");
    11. 				Thread,sleep(1000);
    12.                 return 50;
    13.             }
    14.         });
    15.  
    16.         // NIO线程中异步获取结果
    17.         future.addListener(new GenericFutureListenersuper Integer>>() {
    18.             @Override
    19.             public void operationComplete(Futuresuper Integer> future) throws Exception {
    20.                 log.debug("接收结果:{}",future.getNow());
    21.             }
    22.         });
    23.     }
    24. }

            Netty中的Future对象，可以通过EventLoop的sumbit()方法得到

    • 可以通过Future对象的get方法，阻塞地获取返回结果
    • 也可以通过getNow方法，获取结果，若还没有结果，则返回null，该方法是非阻塞的
    • 还可以通过future.addListener方法，在Callable方法执行的线程中，异步获取返回结果

            Netty Promise

            Promise相当于一个容器，可以用于存放各个线程中的结果，然后让其他线程去获取该结果，这就类似消息队列的功能。

    1. public class NettyPromise {
    2.     public static void main(String[] args) throws ExecutionException, InterruptedException {
    3.         // 创建EventLoop
    4.         EventLoop eventLoop = new NioEventLoopGroup().next();
    5.  
    6.         // 创建Promise对象，用于存放结果
    7.         DefaultPromise promise = new DefaultPromise<>(eventLoop);
    8.  
    9.         new Thread(()->{
    10. 			// 任意一个线程执行计算，计算完毕后向promise中填充结果
    11. 			log.debug("开始计算");
    12.             try {...} catch (InterruptedException e) {
    13.                 e.printStackTrace();
    14.             }
    15.             // 自定义线程向Promise中存放结果
    16.             promise.setSuccess(50);
    17.         }).start();
    18.  
    19. 		// 接收结果的线程
    20.         log.debug("等待结果");
    21. 		log.debug("结果是:{}",promise.get());
    22.     }
    23. }

     四、Handler与Pipeline

            ChannelPipeline是Netty处理请求的责任链，ChannelHandler则是具体处理请求的处理器。实际上每一个channel都有一个处理器的流水线。

            ChannelHandler用来处理Channel.上的各种事件,分为入站、出站两种。

    • 入站处理器通常是ChannellnboundHandlerAdapter的子类，主要用来读取客户端数据，写回结果
    • 出站处理器通常是ChannelOutboundHandlerAdapter的子类，主要对写回结果进行加工

            所有ChannelHandler被连成一串,这就是Pipeline。打个比喻，每个Channel是一个产品的加工车间，Pipeline 是车间中的流水线，ChannelHandler 就是流水线上的各道工序,而后面要讲的ByteBuf是原材料,经过很多工序的加工:先经过一道道入站工序,再经过一道道出站工序最终变成产品。

            Pipeline

            pipeline相当于处理器的容器。初始化channel时，把channelHandler按顺序装在pipeline中，就可以实现按序执行channelHandler了。

            在一个Channel中，只有一个ChannelPipeline。该pipeline在Channel被创建的时候创建。ChannelPipeline包含了一个ChannelHander形成的列表，且所有ChannelHandler都会注册到ChannelPipeline中。

    1. public class PipeLineServer {
    2.     public static void main(String[] args) {
    3.         new ServerBootstrap()
    4.                 .group(new NioEventLoopGroup())
    5.                 .channel(NioServerSocketChannel.class)
    6.                 .childHandler(new ChannelInitializer() {
    7.                     @Override
    8.                     protected void initChannel(SocketChannel socketChannel) throws Exception {
    9.                         // 在socketChannel的pipeline中添加handler
    10.                         // pipeline中handler是带有head与tail节点的双向链表，的实际结构为
    11.     				 	// head <-> handler1 <-> ... <-> handler4 <->tail
    12.                         // Inbound主要处理入站操作，一般为读操作，发生入站操作时会触发Inbound方法
    13.                         // 入站时，handler是从head向后调用的
    14.                         socketChannel.pipeline().addLast("handler1" ,new ChannelInboundHandlerAdapter() {
    15.                             @Override
    16.                             public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    17.                                 System.out.println(Thread.currentThread().getName() + " Inbound handler 1");
    18.                                 // 父类该方法内部会调用fireChannelRead
    19.                                 // 将数据传递给下一个handler
    20.                                 super.channelRead(ctx, msg);
    21.                             }
    22.                         });
    23.                         socketChannel.pipeline().addLast("handler2", new ChannelInboundHandlerAdapter() {
    24.                             @Override
    25.                             public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    26.                                 System.out.println(Thread.currentThread().getName() + " Inbound handler 2");
    27.                                 // 执行write操作，使得Outbound的方法能够得到调用
    28.           socketChannel.writeAndFlush(ctx.alloc().buffer().writeBytes("Server...".getBytes(StandardCharsets.UTF_8)));
    29.                                 super.channelRead(ctx, msg);
    30.                             }
    31.                         });
    32.                         // Outbound主要处理出站操作，一般为写操作，发生出站操作时会触发Outbound方法
    33.                         // 出站时，handler的调用是从tail向前调用的
    34.                         socketChannel.pipeline().addLast("handler3" ,new ChannelOutboundHandlerAdapter(){
    35.                             @Override
    36.                             public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    37.                                 System.out.println(Thread.currentThread().getName() + " Outbound handler 1");
    38.                                 super.write(ctx, msg, promise);
    39.                             }
    40.                         });
    41.                         socketChannel.pipeline().addLast("handler4" ,new ChannelOutboundHandlerAdapter(){
    42.                             @Override
    43.                             public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    44.                                 System.out.println(Thread.currentThread().getName() + " Outbound handler 2");
    45.                                 super.write(ctx, msg, promise);
    46.                             }
    47.                         });
    48.                     }
    49.                 })
    50.                 .bind(8080);
    51.     }
    52. }

             pipeline是结构是一个带有head与tail指针的双向链表，其中的节点为handler，要通过ctx.fireChannelRead(msg)等方法，将当前handler的处理结果传递给下一个handler。

    • 当有入站（Inbound）操作时，会从head开始向后调用handler，直到handler不是处理Inbound操作为止
    • 当有出站（Outbound）操作时，会从tail开始向前调用handler，直到handler不是处理Outbound操作为止

            具体结构如下

             调用顺序如下

            OutboundHandler

            出站处理器ChannelOutboundHandler常用的事件有端口绑定bind、连接服务端connect、写事件write、刷新时间flush、读事件read、主动断开连接disconnect、关闭channel事件close。

            出站处理器中的消息流转则主要有socketChannel.writeAndFlush()与ctx.writeAndFlush()。

            socketChannel.writeAndFlush()

            当handler中调用该方法进行写操作时，会触发Outbound操作，此时是从tail向前寻找OutboundHandler

            ctx.writeAndFlush()

            当handler中调用该方法进行写操作时，会触发Outbound操作，此时是从当前handler向前寻找OutboundHandler

     五、ByteBuf

            创建     

            和java NIO一样，ByteBuf是对字节的封装，我们可以使用如下代码池化一个基于直接内存的ByteBuf，初始容量为16。

    ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(16);

            为了方便调试，这里提供一个工具方法，下文我们会使用到。

    1. private static void log(ByteBuf buffer) {
    2.     int length = buffer.readableBytes();
    3.     int rows = length / 16 + (length % 15 == 0 ? 0 : 1) + 4;
    4.     StringBuilder buf = new StringBuilder(rows * 80 * 2)
    5.         .append("read index:").append(buffer.readerIndex())
    6.         .append(" write index:").append(buffer.writerIndex())
    7.         .append(" capacity:").append(buffer.capacity())
    8.         .append(NEWLINE);
    9.     appendPrettyHexDump(buf, buffer);
    10.     System.out.println(buf.toString());
    11. }

            当ByteBuf的容量不够时，便会自动扩容。我们创建一个初始容量为16的ByteBuf，并向其中加入超过初始容量的字符。

    1. public class ByteBufStudy {
    2.     public static void main(String[] args) {
    3.         // 创建ByteBuf
    4.         ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(16);
    5.         ByteBufUtil.log(buffer);
    6.  
    7.         // 向buffer中写入数据
    8.         StringBuilder sb = new StringBuilder();
    9.         for(int i = 0; i < 20; i++) {
    10.             sb.append("a");
    11.         }
    12.         buffer.writeBytes(sb.toString().getBytes(StandardCharsets.UTF_8));
    13.  
    14.         // 查看写入结果
    15.         ByteBufUtil.log(buffer);
    16.     }
    17. }

            运行结果如下

    1. read index:0 write index:0 capacity:16
    2.  
    3. read index:0 write index:20 capacity:64
    4.          +-------------------------------------------------+
    5.          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
    6. +--------+-------------------------------------------------+----------------+
    7. |00000000| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
    8. |00000010| 61 61 61 61                                     |aaaa            |
    9. +--------+-------------------------------------------------+----------------+

             ByteBuf通过ByteBufAllocator选择allocator并调用对应的buffer()方法来创建的，默认使用直接内存作为ByteBuf，容量为256个字节，可以指定初始容量的大小。

            当ByteBuf的容量无法容纳所有数据时，ByteBuf会进行扩容操作。

            如果在handler中创建ByteBuf，建议使用ChannelHandlerContext ctx.alloc().buffer()来创建

            直接内存与堆内存

            通过该方法创建的ByteBuf，使用的是基于直接内存的ByteBuf

    ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(16);

             可以使用下面的代码来创建池化基于堆的 ByteBuf

    ByteBuf buffer = ByteBufAllocator.DEFAULT.heapBuffer(16);

            也可以使用下面的代码来创建池化基于直接内存的 ByteBuf

    ByteBuf buffer = ByteBufAllocator.DEFAULT.directBuffer(16);
    • 直接内存创建和销毁的代价昂贵，但读写性能高（少一次内存复制），适合配合池化功能一起用
    • 直接内存对 GC 压力小，因为这部分内存不受 JVM 垃圾回收的管理，但也要注意及时主动释放

            最后我们来验证下创建出的ByteBuf是哪种类型的

    1. public class ByteBufStudy {
    2.     public static void main(String[] args) {
    3.         ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(16);
    4.         System.out.println(buffer.getClass());
    5.  
    6.         buffer = ByteBufAllocator.DEFAULT.heapBuffer(16);
    7.         System.out.println(buffer.getClass());
    8.  
    9.         buffer = ByteBufAllocator.DEFAULT.directBuffer(16);
    10.         System.out.println(buffer.getClass());
    11.     }
    12. }

            结果如下

    1. // 使用池化的直接内存
    2. class io.netty.buffer.PooledUnsafeDirectByteBuf
    3.     
    4. // 使用池化的堆内存    
    5. class io.netty.buffer.PooledUnsafeHeapByteBuf
    6.     
    7. // 使用池化的直接内存    
    8. class io.netty.buffer.PooledUnsafeDirectByteBuf

             池化与非池化

            和其他资源一样，ByteBuf的创建也是需要消耗不少资源的，于是我们便想到了池化技术（类似数据库连接池、线程池等）。池化技术的最大意义在于可以重用 ByteBuf，优点有

    • 没有池化，则每次都得创建新的 ByteBuf 实例，这个操作对直接内存代价昂贵，就算是堆内存，也会增加 GC 压力
    • 有了池化，则可以重用池中 ByteBuf 实例，并且采用了与 jemalloc 类似的内存分配算法提升分配效率
    • 高并发时，池化功能更节约内存，减少内存溢出的可能

            池化功能是否开启，可以通过下面的系统环境变量来设置（Netty4.1 之前，池化功能还不成熟，默认是非池化实现）

    -Dio.netty.allocator.type={unpooled|pooled}

            ByteBuf组成

            ByteBuf主要有以下几个组成部分：

            最大容量与当前容量

            在构造ByteBuf时，可传入两个参数，分别代表初始容量和最大容量，若未传入第二个参数（最大容量），最大容量默认为Integer.MAX_VALUE。

            当ByteBuf容量无法容纳所有数据时，会进行扩容操作，若超出最大容量，会抛出java.lang.IndexOutOfBoundsException异常。

            读指针和写指针

            读写操作不同于ByteBuffer只用position进行控制，ByteBuf分别由读指针和写指针两个指针控制。进行读写操作时，无需进行模式的切换。

    • 读指针前的部分被称为废弃部分，是已经读过的内容
    • 读指针与写指针之间的空间称为可读部分
    • 写指针与当前容量之间的空间称为可写部分

             读写

            主要的写方法展示如下

    1. public class ByteBufStudy {
    2.     public static void main(String[] args) {
    3.         // 创建ByteBuf
    4.         ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(16, 20);
    5.         ByteBufUtil.log(buffer);
    6.  
    7.         // 向buffer中写入数据
    8.         buffer.writeBytes(new byte[]{1, 2, 3, 4});
    9.         ByteBufUtil.log(buffer);
    10.  
    11.         // 写入一个数值
    12.         buffer.writeInt(5);
    13.         ByteBufUtil.log(buffer);
    14.  
    15.         // 和writeInt一样，但是小端写入
    16.         buffer.writeIntLE(6);
    17.         ByteBufUtil.log(buffer);
    18.  
    19.         buffer.writeLong(7);
    20.         ByteBufUtil.log(buffer);
    21.     }
    22. }

            结果如下

    1. read index:0 write index:0 capacity:16
    2.  
    3. read index:0 write index:4 capacity:16
    4.          +-------------------------------------------------+
    5.          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
    6. +--------+-------------------------------------------------+----------------+
    7. |00000000| 01 02 03 04                                     |....            |
    8. +--------+-------------------------------------------------+----------------+
    9.  
    10. read index:0 write index:8 capacity:16
    11.          +-------------------------------------------------+
    12.          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
    13. +--------+-------------------------------------------------+----------------+
    14. |00000000| 01 02 03 04 00 00 00 05                         |........        |
    15. +--------+-------------------------------------------------+----------------+
    16.  
    17. read index:0 write index:12 capacity:16
    18.          +-------------------------------------------------+
    19.          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
    20. +--------+-------------------------------------------------+----------------+
    21. |00000000| 01 02 03 04 00 00 00 05 06 00 00 00             |............    |
    22. +--------+-------------------------------------------------+----------------+
    23.  
    24. read index:0 write index:20 capacity:20
    25.          +-------------------------------------------------+
    26.          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
    27. +--------+-------------------------------------------------+----------------+
    28. |00000000| 01 02 03 04 00 00 00 05 06 00 00 00 00 00 00 00 |................|
    29. |00000010| 00 00 00 07                                     |....            |
    30. +--------+-------------------------------------------------+----------------+

             还有一类方法是 set 开头的一系列方法，也可以写入数据，但不会改变写指针位置。

            读取主要是通过一系列read方法进行读取，读取时会根据读取数据的字节数移动读指针，如果需要重复读取，需要调用buffer.markReaderIndex()对读指针进行标记，并通过buffer.resetReaderIndex()将读指针恢复到mark标记的位置。

    1. public class ByteBufStudy {
    2.     public static void main(String[] args) {
    3.         // 创建ByteBuf
    4.         ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(16, 20);
    5.  
    6.         // 向buffer中写入数据
    7.         buffer.writeBytes(new byte[]{1, 2, 3, 4});
    8.         buffer.writeInt(5);
    9.  
    10.         // 读取4个字节
    11.         System.out.println(buffer.readByte());
    12.         System.out.println(buffer.readByte());
    13.         System.out.println(buffer.readByte());
    14.         System.out.println(buffer.readByte());
    15.         ByteBufUtil.log(buffer);
    16.  
    17.         // 通过mark与reset实现重复读取
    18.         buffer.markReaderIndex();
    19.         System.out.println(buffer.readInt());
    20.         ByteBufUtil.log(buffer);
    21.  
    22.         // 恢复到mark标记处
    23.         buffer.resetReaderIndex();
    24.         ByteBufUtil.log(buffer);
    25.     }
    26. }

             结果如下

    1. 1
    2. 2
    3. 3
    4. 4
    5. read index:4 write index:8 capacity:16
    6.          +-------------------------------------------------+
    7.          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
    8. +--------+-------------------------------------------------+----------------+
    9. |00000000| 00 00 00 05                                     |....            |
    10. +--------+-------------------------------------------------+----------------+
    11. 5
    12. read index:8 write index:8 capacity:16
    13.  
    14. read index:4 write index:8 capacity:16
    15.          +-------------------------------------------------+
    16.          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
    17. +--------+-------------------------------------------------+----------------+
    18. |00000000| 00 00 00 05                                     |....            |
    19. +--------+-------------------------------------------------+----------------+

            还有以 get 开头的一系列方法，这些方法不会改变读指针的位置。 

            扩容

            当ByteBuf中的容量无法容纳写入的数据时，会进行扩容操作

    1. buffer.writeLong(7);
    2. ByteBufUtil.log(buffer);

             扩容前后的效果展示如下：

    1. // 扩容前
    2. read index:0 write index:12 capacity:16
    3. ...
    4.  
    5. // 扩容后
    6. read index:0 write index:20 capacity:20
    7.          +-------------------------------------------------+
    8.          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
    9. +--------+-------------------------------------------------+----------------+
    10. |00000000| 01 02 03 04 00 00 00 05 06 00 00 00 00 00 00 00 |................|
    11. |00000010| 00 00 00 07                                     |....            |
    12. +--------+-------------------------------------------------+----------------+

             扩容规则

    • 如何写入后数据大小未超过 512 字节，则选择下一个 16 的整数倍进行扩容，例如写入后大小为 12 字节，则扩容后 capacity 是 16 字节。
    • 如果写入后数据大小超过 512 字节，则选择下一个 2^n，例如写入后大小为 513 字节，则扩容后 capacity 是 2^10=1024 字节（29=512 已经不够了）。
    • 扩容不能超过 maxCapacity，否则会抛出java.lang.IndexOutOfBoundsException异常。
    1. Exception in thread "main" java.lang.IndexOutOfBoundsException: writerIndex(20) + minWritableBytes(8) exceeds maxCapacity(20): PooledUnsafeDirectByteBuf(ridx: 0, widx: 20, cap: 20/20)
    2. ...

             内存释放

            释放方法

            由于 Netty 中有堆外内存（直接内存）的 ByteBuf 实现，堆外内存最好是手动来释放，而不是等 GC 垃圾回收。

    • UnpooledHeapByteBuf 使用的是 JVM 内存，只需等 GC 回收内存即可
    • UnpooledDirectByteBuf 使用的就是直接内存了，需要特殊的方法来回收内存
    • PooledByteBuf 和它的子类使用了池化机制，需要更复杂的规则来回收内存

            Netty 这里采用了引用计数法来控制回收内存，每个 ByteBuf 都实现了 ReferenceCounted 接口。

    • 每个 ByteBuf 对象的初始计数为 1
    • 调用 release 方法计数减 1，如果计数为 0，ByteBuf 内存被回收
    • 调用 retain 方法计数加 1，表示调用者没用完之前，其它 handler 即使调用了 release 也不会造成回收
    • 当计数为 0 时，底层内存会被回收，这时即使 ByteBuf 对象还在，其各个方法均无法正常使用

            释放规则

            因为 pipeline 的存在，一般需要将 ByteBuf 传递给下一个 ChannelHandler，如果在每个 ChannelHandler 中都去调用 release ，就失去了传递性（如果在这个 ChannelHandler 内这个 ByteBuf 已完成了它的使命，那么便无须再传递）。

            基本规则是，谁是最后使用者，谁负责 release：

            （1）起点，对于 NIO 实现来讲，在 io.netty.channel.nio.AbstractNioByteChannel.NioByteUnsafe.read 方法中首次创建 ByteBuf 放入 pipeline（line 163 pipeline.fireChannelRead(byteBuf)）

            （2）入站 ByteBuf 处理原则

    • 对原始 ByteBuf 不做处理，调用 ctx.fireChannelRead(msg) 向后传递，这时无须 release
    • 将原始 ByteBuf 转换为其它类型的 Java 对象，这时 ByteBuf 就没用了，必须 release
    • 如果不调用 ctx.fireChannelRead(msg) 向后传递，那么也必须 release
    • 注意各种异常，如果 ByteBuf 没有成功传递到下一个 ChannelHandler，必须 release
    • 假设消息一直向后传，那么 TailContext 会负责释放未处理消息（原始的 ByteBuf）

            （3）出站 ByteBuf 处理原则

            出站消息最终都会转为 ByteBuf 输出，一直向前传，由 HeadContext flush 后 release

            （4）异常处理原则

            有时候不清楚 ByteBuf 被引用了多少次，但又必须彻底释放，可以循环调用 release 直到返回 true

    while (!buffer.release()) {}

            当ByteBuf被传到了pipeline的head与tail时，ByteBuf会被其中的方法彻底释放，但前提是ByteBuf被传递到了head与tail中。TailConext中释放ByteBuf的源码如下：

    1. protected void onUnhandledInboundMessage(Object msg) {
    2.     try {
    3.         logger.debug("Discarded inbound message {} that reached at the tail of the pipeline. Please check your pipeline configuration.", msg);
    4.     } finally {
    5.         // 具体的释放方法
    6.         ReferenceCountUtil.release(msg);
    7.     }
    8. }

            判断传过来的是否为ByteBuf，是的话才需要释放

    1. public static boolean release(Object msg) {
    2. 	return msg instanceof ReferenceCounted ? ((ReferenceCounted)msg).release() : false;
    3. }

            切片

             ByteBuf切片是【零拷贝】的体现之一，对原始 ByteBuf 进行切片成多个 ByteBuf，切片后的 ByteBuf 并没有发生内存复制，还是使用原始 ByteBuf 的内存，切片后的 ByteBuf 维护独立的 read，write 指针。

            得到分片后的buffer后，要调用其retain方法，使其内部的引用计数加一。避免原ByteBuf释放，导致切片buffer无法使用。修改原ByteBuf中的值，也会影响切片后得到的ByteBuf。

    1. public class TestSlice {
    2.     public static void main(String[] args) {
    3.         // 创建ByteBuf
    4.         ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(16, 20);
    5.  
    6.         // 向buffer中写入数据
    7.         buffer.writeBytes(new byte[]{1, 2, 3, 4, 5, 6, 7, 8, 9, 10});
    8.  
    9.         // 将buffer分成两部分
    10.         ByteBuf slice1 = buffer.slice(0, 5);
    11.         ByteBuf slice2 = buffer.slice(5, 5);
    12.  
    13.         // 需要让分片的buffer引用计数加一
    14.         // 避免原Buffer释放导致分片buffer无法使用
    15.         slice1.retain();
    16.         slice2.retain();
    17.         
    18.         ByteBufUtil.log(slice1);
    19.         ByteBufUtil.log(slice2);
    20.  
    21.         // 更改原始buffer中的值
    22.         System.out.println("===========修改原buffer中的值===========");
    23.         buffer.setByte(0,5);
    24.  
    25.         System.out.println("===========打印slice1===========");
    26.         ByteBufUtil.log(slice1);
    27.     }
    28. }

             结果如下

    1. read index:0 write index:5 capacity:5
    2.          +-------------------------------------------------+
    3.          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
    4. +--------+-------------------------------------------------+----------------+
    5. |00000000| 01 02 03 04 05                                  |.....           |
    6. +--------+-------------------------------------------------+----------------+
    7. read index:0 write index:5 capacity:5
    8.          +-------------------------------------------------+
    9.          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
    10. +--------+-------------------------------------------------+----------------+
    11. |00000000| 06 07 08 09 0a                                  |.....           |
    12. +--------+-------------------------------------------------+----------------+
    13. ===========修改原buffer中的值===========
    14. ===========打印slice1===========
    15. read index:0 write index:5 capacity:5
    16.          +-------------------------------------------------+
    17.          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
    18. +--------+-------------------------------------------------+----------------+
    19. |00000000| 05 02 03 04 05                                  |.....           |
    20. +--------+-------------------------------------------------+----------------+
  • 相关阅读:
    欲知己之所防，先知彼之所攻——论Hook 技术的攻防对抗
    ElasticSearch实战
    使用.NET简单实现一个Redis的高性能克隆版（七-完结）
    算法-链表-反转链表
    量子计算（十）：量子计算原理
    命令与文件的查找
    HTML+CSS简单的网页制作期末作业——浙江旅游景点介绍网页制作
    Aware依赖注入管理
    编程的简单实例，编程零基础入门教程，中文编程开发语言工具下载
    十七、Rust集成MQTT Client
  • 原文地址：https://blog.csdn.net/wzngzaixiaomantou/article/details/127947714