精通Netty，那倒是把这个8个东西说清楚呀！（）

　　本篇文章为你整理了精通Netty，那倒是把这个8个东西说清楚呀！（）的详细内容，包含有精通Netty，那倒是把这个8个东西说清楚呀！，希望能帮助你了解精通Netty，那倒是把这个8个东西说清楚呀！。

　　1、什么是 Netty

Netty is an asynchronous event-driven network application framework

　　for rapid development of maintainable high performance protocol servers clients.

　　Netty 是一个异步的、基于事件驱动的网络应用框架，用于快速开发可维护、高性能的网络服务器和客户端

　　注意：netty的异步还是基于多路复用的，并没有实现真正意义上的异步IO

　　2、Netty 的优势

　　如果使用传统 NIO，其工作量大，bug 多

　　需要自己构建协议

　　解决 TCP 传输问题，如粘包、半包

　　因为 bug 的存在，epoll 空轮询导致 CPU 100%

　　Netty 对 API 进行增强，使之更易用，如

　　FastThreadLocal = ThreadLocal

　　ByteBuf = ByteBuffer

　　3、入门案例

　　1、服务器端代码

public class HelloServer {

　　 public static void main(String[] args) {

　　 // 1、启动器，负责装配netty组件，启动服务器

　　 new ServerBootstrap()

　　 // 2、创建 NioEventLoopGroup，可以简单理解为线程池 + Selector

　　 .group(new NioEventLoopGroup())

　　 // 3、选择服务器的 ServerSocketChannel 实现

　　 .channel(NioServerSocketChannel.class)

　　 // 4、child 负责处理读写，该方法决定了 child 执行哪些操作

　　 // ChannelInitializer 处理器（仅执行一次）

　　 // 它的作用是待客户端 SocketChannel 建立连接后，执行 initChannel 以便添加更多的处理器

　　 .childHandler(new ChannelInitializer NioSocketChannel () {

　　 @Override

　　 protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {

　　 // 5、SocketChannel的处理器，使用StringDecoder解码，ByteBuf= String

　　 nioSocketChannel.pipeline().addLast(new StringDecoder());

　　 // 6、SocketChannel的业务处理，使用上一个处理器的处理结果

　　 nioSocketChannel.pipeline().addLast(new SimpleChannelInboundHandler String () {

　　 @Override

　　 protected void channelRead0(ChannelHandlerContext channelHandlerContext, String s) throws Exception {

　　 System.out.println(s);

　　 // 7、ServerSocketChannel绑定8080端口

　　 }).bind(8080);

　　2、客户端代码

public class HelloClient {

　　 public static void main(String[] args) throws InterruptedException {

　　 new Bootstrap()

　　 .group(new NioEventLoopGroup())

　　 // 选择客户 Socket 实现类，NioSocketChannel 表示基于 NIO 的客户端实现

　　 .channel(NioSocketChannel.class)

　　 // ChannelInitializer 处理器（仅执行一次）

　　 // 它的作用是待客户端SocketChannel建立连接后，执行initChannel以便添加更多的处理器

　　 .handler(new ChannelInitializer Channel () {

　　 @Override

　　 protected void initChannel(Channel channel) throws Exception {

　　 // 消息会经过通道 handler 处理，这里是将 String = ByteBuf 编码发出

　　 channel.pipeline().addLast(new StringEncoder());

　　 // 指定要连接的服务器和端口

　　 .connect(new InetSocketAddress("localhost", 8080))

　　 // Netty 中很多方法都是异步的，如 connect

　　 // 这时需要使用 sync 方法等待 connect 建立连接完毕

　　 .sync()

　　 // 获取 channel 对象，它即为通道抽象，可以进行数据读写操作

　　 .channel()

　　 // 写入消息并清空缓冲区

　　 .writeAndFlush("hello world");

　　3、运行流程

　　左：客户端右：服务器端

　　组件解释

　　channel 可以理解为数据的通道

　　msg 理解为流动的数据，最开始输入是 ByteBuf，但经过 pipeline 中的各个 handler 加工，会变成其它类型对象，最后输出又变成 ByteBuf

　　handler 可以理解为数据的处理工序

　　
工序有多道，合在一起就是 pipeline（传递途径），pipeline 负责发布事件（读、读取完成…）传播给每个 handler， handler 对自己感兴趣的事件进行处理（重写了相应事件处理方法）

　　pipeline 中有多个 handler，处理时会依次调用其中的 handler

　　
eventLoop 可以管理多个 channel 的 io 操作，并且一旦 eventLoop 负责了某个 channel，就会将其与 channel 进行绑定，以后该 channel 中的 io 操作都由该 eventLoop 负责

　　eventLoop 既可以执行 io 操作，也可以进行任务处理，每个 eventLoop 有自己的任务队列，队列里可以堆放多个 channel 的待处理任务，任务分为普通任务、定时任务

　　eventLoop 按照 pipeline 顺序，依次按照 handler 的规划（代码）处理数据，可以为每个 handler 指定不同的 eventLoop

　　
事件循环对象 EventLoop

　　EventLoop 本质是一个单线程执行器（同时维护了一个 Selector），里面有 run 方法处理一个或多个 Channel 上源源不断的 io 事件

　　它的继承关系如下

　　
事件循环组 EventLoopGroup

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

　　继承自 netty 自己的 EventExecutorGroup

　　实现了 Iterable 接口提供遍历 EventLoop 的能力

　　另有 next 方法获取集合中下一个 EventLoop

　　
public static void main(String[] args) {

　　 // 创建拥有两个EventLoop的NioEventLoopGroup，对应两个线程

　　 EventLoopGroup group = new NioEventLoopGroup(2);

　　 // 通过next方法可以获得下一个 EventLoop

　　 System.out.println(group.next());

　　 // 通过EventLoop执行普通任务

　　 group.next().execute(()- {

　　 System.out.println(Thread.currentThread().getName() + " hello");

　　 // 通过EventLoop执行定时任务

　　 group.next().scheduleAtFixedRate(()- {

　　 System.out.println(Thread.currentThread().getName() + " hello2");

　　 }, 0, 1, TimeUnit.SECONDS);

　　 // 优雅地关闭

　　 group.shutdownGracefully();

　　输出结果如下

io.netty.channel.nio.NioEventLoop@7bb11784

　　io.netty.channel.nio.NioEventLoop@33a10788

　　nioEventLoopGroup-2-1 hello

　　nioEventLoopGroup-2-2 hello2

　　关闭 EventLoopGroup

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

　　1.2 处理 IO 任务

　　服务器代码

public class MyServer {

　　 public static void main(String[] args) {

　　 new ServerBootstrap()

　　 .group(new NioEventLoopGroup())

　　 .channel(NioServerSocketChannel.class)

　　 .childHandler(new ChannelInitializer SocketChannel () {

　　 @Override

　　 protected void initChannel(SocketChannel socketChannel) throws Exception {

　　 socketChannel.pipeline().addLast(new ChannelInboundHandlerAdapter() {

　　 @Override

　　 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

　　 ByteBuf buf = (ByteBuf) msg;

　　 System.out.println(Thread.currentThread().getName() + " " + buf.toString(StandardCharsets.UTF_8));

　　 .bind(8080);

　　客户端代码

public class MyClient {

　　 public static void main(String[] args) throws IOException, InterruptedException {

　　 Channel channel = new Bootstrap()

　　 .group(new NioEventLoopGroup())

　　 .channel(NioSocketChannel.class)

　　 .handler(new ChannelInitializer SocketChannel () {

　　 @Override

　　 protected void initChannel(SocketChannel socketChannel) throws Exception {

　　 socketChannel.pipeline().addLast(new StringEncoder());

　　 .connect(new InetSocketAddress("localhost", 8080))

　　 .sync()

　　 .channel();

　　 System.out.println(channel);

　　 // 此处打断点调试，调用 channel.writeAndFlush(...);

　　 System.in.read();

　　1.3 分工

　　Bootstrap 的 group () 方法可以传入两个 EventLoopGroup 参数，分别负责处理不同的事件

public class MyServer {

　　 public static void main(String[] args) {

　　 new ServerBootstrap()

　　 // 两个Group，分别为Boss 负责Accept事件，Worker 负责读写事件

　　 .group(new NioEventLoopGroup(1), new NioEventLoopGroup(2))

　　多个客户端分别发送 hello 结果

nioEventLoopGroup-3-1 hello1

　　nioEventLoopGroup-3-2 hello2

　　nioEventLoopGroup-3-1 hello3

　　nioEventLoopGroup-3-2 hello4

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

　　增加自定义 EventLoopGroup

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

 public class MyServer {

　　 public static void main(String[] args) {

　　 // 增加自定义的非NioEventLoopGroup

　　 EventLoopGroup group = new DefaultEventLoopGroup();

　　 new ServerBootstrap()

　　 .group(new NioEventLoopGroup(1), new NioEventLoopGroup(2))

　　 .channel(NioServerSocketChannel.class)

　　 .childHandler(new ChannelInitializer SocketChannel () {

　　 @Override

　　 protected void initChannel(SocketChannel socketChannel) throws Exception {

　　 // 增加两个handler，第一个使用NioEventLoopGroup处理，第二个使用自定义EventLoopGroup处理

　　 socketChannel.pipeline().addLast("nioHandler",new ChannelInboundHandlerAdapter() {

　　 @Override

　　 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

　　 ByteBuf buf = (ByteBuf) msg;

　　 System.out.println(Thread.currentThread().getName() + " " + buf.toString(StandardCharsets.UTF_8));

　　 // 调用下一个handler

　　 ctx.fireChannelRead(msg);

　　 // 该handler绑定自定义的Group

　　 .addLast(group, "myHandler", new ChannelInboundHandlerAdapter() {

　　 @Override

　　 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

　　 ByteBuf buf = (ByteBuf) msg;

　　 System.out.println(Thread.currentThread().getName() + " " + buf.toString(StandardCharsets.UTF_8));

　　 .bind(8080);

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

　　切换的实现

　　不同的 EventLoopGroup 切换的实现原理如下

　　由上面的图可以看出，当 handler 中绑定的 Group 不同时，需要切换 Group 来执行不同的任务

static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {

　　 final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next);

　　 // 获得下一个EventLoop, excutor 即为 EventLoopGroup

　　 EventExecutor executor = next.executor();

　　 // 如果下一个EventLoop 在当前的 EventLoopGroup中

　　 if (executor.inEventLoop()) {

　　 // 使用当前 EventLoopGroup 中的 EventLoop 来处理任务

　　 next.invokeChannelRead(m);

　　 } else {

　　 // 否则让另一个 EventLoopGroup 中的 EventLoop 来创建任务并执行

　　 executor.execute(new Runnable() {

　　 public void run() {

　　 next.invokeChannelRead(m);

　　如果两个 handler 绑定的是同一个 EventLoopGroup，那么就直接调用

　　否则，把要调用的代码封装为一个任务对象，由下一个 handler 的 EventLoopGroup 来调用

　　2、Channel

　　Channel 的常用方法

　　close () 可以用来关闭 Channel

　　closeFuture () 用来处理 Channel 的关闭

　　sync 方法作用是同步等待 Channel 关闭

　　而 addListener 方法是异步等待 Channel 关闭

　　
因为缓冲机制，数据被写入到 Channel 中以后，不会立即被发送

　　只有当缓冲满了或者调用了 flush () 方法后，才会将数据通过 Channel 发送出去

public class MyClient {

　　 public static void main(String[] args) throws IOException, InterruptedException {

　　 ChannelFuture channelFuture = new Bootstrap()

　　 .group(new NioEventLoopGroup())

　　 .channel(NioSocketChannel.class)

　　 .handler(new ChannelInitializer SocketChannel () {

　　 @Override

　　 protected void initChannel(SocketChannel socketChannel) throws Exception {

　　 socketChannel.pipeline().addLast(new StringEncoder());

　　 // 该方法为异步非阻塞方法，主线程调用后不会被阻塞，真正去执行连接操作的是NIO线程

　　 // NIO线程：NioEventLoop 中的线程

　　 .connect(new InetSocketAddress("localhost", 8080));

　　 // 该方法用于等待连接真正建立

　　 channelFuture.sync();

　　 // 获取客户端-服务器之间的Channel对象

　　 Channel channel = channelFuture.channel();

　　 channel.writeAndFlush("hello world");

　　 System.in.read();

　　如果我们去掉 channelFuture.sync() 方法，会服务器无法收到 hello world

　　这是因为建立连接 (connect) 的过程是异步非阻塞的，若不通过 sync() 方法阻塞主线程，等待连接真正建立，这时通过 channelFuture.channel () 拿到的 Channel 对象，并不是真正与服务器建立好连接的 Channel，也就没法将信息正确的传输给服务器端

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

　　下面还有一种方法，用于异步获取建立连接后的 Channel 和发送数据，使得执行这些操作的线程是 NIO 线程（去执行 connect 操作的线程）

　　addListener 方法

　　通过这种方法可以在 NIO 线程中获取 Channel 并发送数据，而不是在主线程中执行这些操作

public class MyClient {

　　 public static void main(String[] args) throws IOException, InterruptedException {

　　 ChannelFuture channelFuture = new Bootstrap()

　　 .group(new NioEventLoopGroup())

　　 .channel(NioSocketChannel.class)

　　 .handler(new ChannelInitializer SocketChannel () {

　　 @Override

　　 protected void initChannel(SocketChannel socketChannel) throws Exception {

　　 socketChannel.pipeline().addLast(new StringEncoder());

　　 // 该方法为异步非阻塞方法，主线程调用后不会被阻塞，真正去执行连接操作的是NIO线程

　　 // NIO线程：NioEventLoop 中的线程

　　 .connect(new InetSocketAddress("localhost", 8080));

　　 // 当connect方法执行完毕后，也就是连接真正建立后

　　 // 会在NIO线程中调用operationComplete方法

　　 channelFuture.addListener(new ChannelFutureListener() {

　　 @Override

　　 public void operationComplete(ChannelFuture channelFuture) throws Exception {

　　 Channel channel = channelFuture.channel();

　　 channel.writeAndFlush("hello world");

　　 System.in.read();

public class ReadClient {

　　 public static void main(String[] args) throws InterruptedException {

　　 // 创建EventLoopGroup，使用完毕后关闭

　　 NioEventLoopGroup group = new NioEventLoopGroup();

　　 ChannelFuture channelFuture = new Bootstrap()

　　 .group(group)

　　 .channel(NioSocketChannel.class)

　　 .handler(new ChannelInitializer SocketChannel () {

　　 @Override

　　 protected void initChannel(SocketChannel socketChannel) throws Exception {

　　 socketChannel.pipeline().addLast(new StringEncoder());

　　 .connect(new InetSocketAddress("localhost", 8080));

　　 channelFuture.sync();

　　 Channel channel = channelFuture.channel();

　　 Scanner scanner = new Scanner(System.in);

　　 // 创建一个线程用于输入并向服务器发送

　　 new Thread(()- {

　　 while (true) {

　　 String msg = scanner.next();

　　 if ("q".equals(msg)) {

　　 // 关闭操作是异步的，在NIO线程中执行

　　 channel.close();

　　 break;

　　 channel.writeAndFlush(msg);

　　 }, "inputThread").start();

　　 // 获得closeFuture对象

　　 ChannelFuture closeFuture = channel.closeFuture();

　　 System.out.println("waiting close...");

　　 // 同步等待NIO线程执行完close操作

　　 closeFuture.sync();

　　 // 关闭之后执行一些操作，可以保证执行的操作一定是在channel关闭以后执行的

　　 System.out.println("关闭之后执行一些额外操作...");

　　 // 关闭EventLoopGroup

　　 group.shutdownGracefully();

　　关闭channel

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

　　如果我们想在 channel 真正关闭以后，执行一些额外的操作，可以选择以下两种方法来实现

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

// 获得closeFuture对象

　　ChannelFuture closeFuture = channel.closeFuture();

　　// 同步等待NIO线程执行完close操作

　　closeFuture.sync();

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

closeFuture.addListener(new ChannelFutureListener() {

　　 @Override

　　 public void operationComplete(ChannelFuture channelFuture) throws Exception {

　　 // 等待channel关闭后才执行的操作

　　 System.out.println("关闭之后执行一些额外操作...");

　　 // 关闭EventLoopGroup

　　 group.shutdownGracefully();

　　3、Future 与 Promise

　　3.1 概念

　　netty 中的 Future 与 jdk 中的 Future 同名，但是是两个接口

　　netty 的 Future 继承自 jdk 的 Future，而 Promise 又对 netty Future 进行了扩展

　　
netty Future 可以同步等待任务结束得到结果，也可以异步方式得到结果，但都是要等任务结束

　　
netty Promise 不仅有 netty Future 的功能，而且脱离了任务独立存在，只作为两个线程间传递结果的容器

　　3.2 JDK Future

public class JdkFuture {

　　 public static void main(String[] args) throws ExecutionException, InterruptedException {

　　 ThreadFactory factory = new ThreadFactory() {

　　 @Override

　　 public Thread newThread(Runnable r) {

　　 return new Thread(r, "JdkFuture");

　　 // 创建线程池

　　 ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10,10, TimeUnit.SECONDS, new ArrayBlockingQueue (10), factory);

　　 // 获得Future对象

　　 Future Integer future = executor.submit(new Callable Integer () {

　　 @Override

　　 public Integer call() throws Exception {

　　 TimeUnit.SECONDS.sleep(1);

　　 return 50;

　　 // 通过阻塞的方式，获得运行结果

　　 System.out.println(future.get());

　　3.3 Netty Future

public class NettyFuture {

　　 public static void main(String[] args) throws ExecutionException, InterruptedException {

　　 NioEventLoopGroup group = new NioEventLoopGroup();

　　 // 获得 EventLoop 对象

　　 EventLoop eventLoop = group.next();

　　 Future Integer future = eventLoop.submit(new Callable Integer () {

　　 @Override

　　 public Integer call() throws Exception {

　　 return 50;

　　 // 主线程中获取结果

　　 System.out.println(Thread.currentThread().getName() + " 获取结果");

　　 System.out.println("getNow " + future.getNow());

　　 System.out.println("get " + future.get());

　　 // NIO线程中异步获取结果

　　 future.addListener(new GenericFutureListener Future ? super Integer () {

　　 @Override

　　 public void operationComplete(Future ? super Integer future) throws Exception {

　　 System.out.println(Thread.currentThread().getName() + " 获取结果");

　　 System.out.println("getNow " + future.getNow());

　　运行结果

main 获取结果

　　getNow null

　　get 50

　　nioEventLoopGroup-2-1 获取结果

　　getNow 50

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

　　
还可以通过 future.addListener 方法，在 Callable 方法执行的线程中，异步获取返回结果

　　
3.4 Netty Promise

　　Promise 相当于一个容器，可以用于存放各个线程中的结果，然后让其他线程去获取该结果

public class NettyPromise {

　　 public static void main(String[] args) throws ExecutionException, InterruptedException {

　　 // 创建EventLoop

　　 NioEventLoopGroup group = new NioEventLoopGroup();

　　 EventLoop eventLoop = group.next();

　　 // 创建Promise对象，用于存放结果

　　 DefaultPromise Integer promise = new DefaultPromise (eventLoop);

　　 new Thread(()- {

　　 try {

　　 TimeUnit.SECONDS.sleep(1);

　　 } catch (InterruptedException e) {

　　 e.printStackTrace();

　　 // 自定义线程向Promise中存放结果

　　 promise.setSuccess(50);

　　 }).start();

　　 // 主线程从Promise中获取结果

　　 System.out.println(Thread.currentThread().getName() + " " + promise.get());

　　4、Handler 与 Pipeline

　　4.1 Pipeline

public class PipeLineServer {

　　 public static void main(String[] args) {

　　 new ServerBootstrap()

　　 .group(new NioEventLoopGroup())

　　 .channel(NioServerSocketChannel.class)

　　 .childHandler(new ChannelInitializer SocketChannel () {

　　 @Override

　　 protected void initChannel(SocketChannel socketChannel) throws Exception {

　　 // 在socketChannel的pipeline中添加handler

　　 // pipeline中handler是带有head与tail节点的双向链表，的实际结构为

　　 // head - handler1 - ... - handler4 - tail

　　 // Inbound主要处理入站操作，一般为读操作，发生入站操作时会触发Inbound方法

　　 // 入站时，handler是从head向后调用的

　　 socketChannel.pipeline().addLast("handler1" ,new ChannelInboundHandlerAdapter() {

　　 @Override

　　 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

　　 System.out.println(Thread.currentThread().getName() + " Inbound handler 1");

　　 // 父类该方法内部会调用fireChannelRead

　　 // 将数据传递给下一个handler

　　 super.channelRead(ctx, msg);

　　 socketChannel.pipeline().addLast("handler2", new ChannelInboundHandlerAdapter() {

　　 @Override

　　 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

　　 System.out.println(Thread.currentThread().getName() + " Inbound handler 2");

　　 // 执行write操作，使得Outbound的方法能够得到调用

　　 socketChannel.writeAndFlush(ctx.alloc().buffer().writeBytes("Server...".getBytes(StandardCharsets.UTF_8)));

　　 super.channelRead(ctx, msg);

　　 // Outbound主要处理出站操作，一般为写操作，发生出站操作时会触发Outbound方法

　　 // 出站时，handler的调用是从tail向前调用的

　　 socketChannel.pipeline().addLast("handler3" ,new ChannelOutboundHandlerAdapter(){

　　 @Override

　　 public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {

　　 System.out.println(Thread.currentThread().getName() + " Outbound handler 1");

　　 super.write(ctx, msg, promise);

　　 socketChannel.pipeline().addLast("handler4" ,new ChannelOutboundHandlerAdapter(){

　　 @Override

　　 public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {

　　 System.out.println(Thread.currentThread().getName() + " Outbound handler 2");

　　 super.write(ctx, msg, promise);

　　 .bind(8080);

　　运行结果如下

nioEventLoopGroup-2-2 Inbound handler 1

　　nioEventLoopGroup-2-2 Inbound handler 2

　　nioEventLoopGroup-2-2 Outbound handler 2

　　nioEventLoopGroup-2-2 Outbound handler 1

　　通过 channel.pipeline ().addLast (name, handler) 添加 handler 时，记得给 handler 取名字。这样可以调用 pipeline 的 addAfter、addBefore 等方法更灵活地向 pipeline 中添加 handler

　　handler 需要放入通道的 pipeline 中，才能根据放入顺序来使用 handler

　　pipeline 是结构是一个带有 head 与 tail 指针的双向链表，其中的节点为 handler

　　要通过 ctx.fireChannelRead (msg) 等方法，将当前 handler 的处理结果传递给下一个 handler

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

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

　　具体结构如下

　　调用顺序如下

　　4.2 OutboundHandler

　　socketChannel.writeAndFlush()

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

　　ctx.writeAndFlush()

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

　　4.3 EmbeddedChannel

　　EmbeddedChannel 可以用于测试各个 handler，通过其构造函数按顺序传入需要测试 handler，然后调用对应的 Inbound 和 Outbound 方法即可

public class TestEmbeddedChannel {

　　 public static void main(String[] args) {

　　 ChannelInboundHandlerAdapter h1 = new ChannelInboundHandlerAdapter() {

　　 @Override

　　 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

　　 System.out.println("1");

　　 super.channelRead(ctx, msg);

　　 ChannelInboundHandlerAdapter h2 = new ChannelInboundHandlerAdapter() {

　　 @Override

　　 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

　　 System.out.println("2");

　　 super.channelRead(ctx, msg);

　　 ChannelOutboundHandlerAdapter h3 = new ChannelOutboundHandlerAdapter() {

　　 @Override

　　 public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {

　　 System.out.println("3");

　　 super.write(ctx, msg, promise);

　　 ChannelOutboundHandlerAdapter h4 = new ChannelOutboundHandlerAdapter() {

　　 @Override

　　 public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {

　　 System.out.println("4");

　　 super.write(ctx, msg, promise);

　　 // 用于测试Handler的Channel

　　 EmbeddedChannel channel = new EmbeddedChannel(h1, h2, h3, h4);

　　 // 执行Inbound操作

　　 channel.writeInbound(ByteBufAllocator.DEFAULT.buffer().writeBytes("hello".getBytes(StandardCharsets.UTF_8)));

　　 // 执行Outbound操作

　　 channel.writeOutbound(ByteBufAllocator.DEFAULT.buffer().writeBytes("hello".getBytes(StandardCharsets.UTF_8)));

　　5、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());

　　该方法可以帮助我们更为详细地查看 ByteBuf 中的内容

　　5.1 创建

public class ByteBufStudy {

　　 public static void main(String[] args) {

　　 // 创建ByteBuf

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

　　 ByteBufUtil.log(buffer);

　　 // 向buffer中写入数据

　　 StringBuilder sb = new StringBuilder();

　　 for(int i = 0; i i++) {

　　 sb.append("a");

　　 buffer.writeBytes(sb.toString().getBytes(StandardCharsets.UTF_8));

　　 // 查看写入结果

　　 ByteBufUtil.log(buffer);

　　运行结果

read index:0 write index:0 capacity:16

　　read index:0 write index:20 capacity:64

　　 +-------------------------------------------------+

　　 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 aaaa

　　+--------+-------------------------------------------------+----------------+

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

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

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

　　5.2 直接内存与堆内存

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

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

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

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

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

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

　　直接内存创建和销毁的代价昂贵，但读写性能高（少一次内存复制），适合配合池化功能一起用

　　直接内存对 GC 压力小，因为这部分内存不受 JVM 垃圾回收的管理，但也要注意及时主动释放

　　验证

public class ByteBufStudy {

　　 public static void main(String[] args) {

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

　　 System.out.println(buffer.getClass());

　　 buffer = ByteBufAllocator.DEFAULT.heapBuffer(16);

　　 System.out.println(buffer.getClass());

　　 buffer = ByteBufAllocator.DEFAULT.directBuffer(16);

　　 System.out.println(buffer.getClass());

　　// 使用池化的直接内存

　　class io.netty.buffer.PooledUnsafeDirectByteBuf

　　// 使用池化的堆内存

　　class io.netty.buffer.PooledUnsafeHeapByteBuf

　　// 使用池化的直接内存

　　class io.netty.buffer.PooledUnsafeDirectByteBuf

　　5.3 池化与非池化

　　池化的最大意义在于可以重用 ByteBuf，优点有

　　没有池化，则每次都得创建新的 ByteBuf 实例，这个操作对直接内存代价昂贵，就算是堆内存，也会增加 GC 压力

　　有了池化，则可以重用池中 ByteBuf 实例，并且采用了与 jemalloc 类似的内存分配算法提升分配效率

　　高并发时，池化功能更节约内存，减少内存溢出的可能

　　池化功能是否开启，可以通过下面的系统环境变量来设置

-Dio.netty.allocator.type={unpooledpooled}

　　4.1 以后，非 Android 平台默认启用池化实现，Android 平台启用非池化实现

　　4.1 之前，池化功能还不成熟，默认是非池化实现

　　5.4 组成

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

　　最大容量与当前容量

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

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

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

　　读指针前的部分被称为废弃部分，是已经读过的内容

　　5.5 写入

　　常用方法如下

　　注意

　　这些方法的未指明返回值的，其返回值都是 ByteBuf，意味着可以链式调用来写入不同的数据

　　网络传输中，默认习惯是 Big Endian，使用 writeInt (int value)

　　使用方法

public class ByteBufStudy {

　　 public static void main(String[] args) {

　　 // 创建ByteBuf

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

　　 ByteBufUtil.log(buffer);

　　 // 向buffer中写入数据

　　 buffer.writeBytes(new byte[]{1, 2, 3, 4});

　　 ByteBufUtil.log(buffer);

　　 buffer.writeInt(5);

　　 ByteBufUtil.log(buffer);

　　 buffer.writeIntLE(6);

　　 ByteBufUtil.log(buffer);

　　 buffer.writeLong(7);

　　 ByteBufUtil.log(buffer);

　　运行结果

read index:0 write index:0 capacity:16

　　read index:0 write index:4 capacity:16

　　 +-------------------------------------------------+

　　 0 1 2 3 4 5 6 7 8 9 a b c d e f

　　+--------+-------------------------------------------------+----------------+

　　00000000 01 02 03 04 ....

　　+--------+-------------------------------------------------+----------------+

　　read index:0 write index:8 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 ........

　　+--------+-------------------------------------------------+----------------+

　　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 06 00 00 00 ............

　　+--------+-------------------------------------------------+----------------+

　　read index:0 write index:20 capacity:20

　　 +-------------------------------------------------+

　　 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 06 00 00 00 00 00 00 00 ................

　　00000010 00 00 00 07 ....

　　+--------+-------------------------------------------------+----------------+

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

　　5.6 扩容

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

buffer.writeLong(7);

　　ByteBufUtil.log(buffer);

　　// 扩容前

　　read index:0 write index:12 capacity:16

　　// 扩容后

　　read index:0 write index:20 capacity:20

　　 +-------------------------------------------------+

　　 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 06 00 00 00 00 00 00 00 ................

　　00000010 00 00 00 07 ....

　　+--------+-------------------------------------------------+----------------+

　　
例如写入后大小为 513 字。

郑重声明：本文由网友发布，不代表盛行IT的观点，版权归原作者所有，仅为传播更多信息之目的，如有侵权请联系，我们将第一时间修改或删除，多谢。

相关文章阅读