一、概述

1. 什么是Netty

1. 简介

内容参考：https://www.jianshu.com/p/b9f3f6a16911
官网：https://netty.io/index.html

Netty 是一个利用 Java 的高级网络的能力，隐藏其背后的复杂性而提供一个易于使用的 API 的客户端/服务器框架。
Netty 是一个广泛使用的 Java 网络编程框架（Netty 在 2011 年获得了Duke’s Choice Award，见 https://www.java.net/dukeschoice/2011）。它活跃和成长于用户社区，像大型公司 Facebook 和 Instagram 以及流行开源项目如 Infinispan, HornetQ, Vert.x, Apache Cassandra 和 Elasticsearch 等，都利用其强大的对于网络抽象的核心代码。

Netty is an asynchronous event-driven network application framework
for rapid development of maintainable high performance protocol servers & clients.

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

Netty作者

2. 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 - 分布式协调框架

3. Netty的优势

Netty vs NIO，工作量大，bug 多

需要自己构建协议
解决 TCP 传输问题，如粘包、半包
epoll 空轮询导致 CPU 100%
对 API 进行增强，使之更易用，如 FastThreadLocal => ThreadLocal，ByteBuf => ByteBuffer

Netty vs 其它网络应用框架

Mina 由 apache 维护，将来 3.x 版本可能会有较大重构，破坏 API 向下兼容性，Netty 的开发迭代更迅速，API 更简洁、文档更优秀
久经考验，16年，Netty 版本
- 2.x 2004
- 3.x 2008
- 4.x 2013
- 5.x 已废弃（没有明显的性能提升，维护成本高）

2. Netty和Tomcat有什么区别

Netty和Tomcat最大的区别就在于通信协议，Tomcat是基于Http协议的，他的实质是一个基于http协议的web容器，但是Netty不一样，他能通过编程自定义各种协议，因为netty能够通过codec自己来编码/解码字节流，完成类似redis访问的功能，这就是netty和tomcat最大的不同。

有人说netty的性能就一定比tomcat性能高，其实不然，tomcat从6.x开始就支持了nio模式，并且后续还有APR模式——一种通过jni调用apache网络库的模式，相比于旧的bio模式，并发性能得到了很大提高，特别是APR模式，而netty是否比tomcat性能更高，则要取决于netty程序作者的技术实力了

3. 为什么Netty受大众欢迎

如第一部分所述，netty是一款收到大公司青睐的框架，在我看来，netty能够受到青睐的原因有三：

并发高
传输快
封装好

4. Netty为什么并发高

Netty是一款基于NIO（Nonblocking I/O，非阻塞IO）开发的网络通信框架，对比于BIO（Blocking I/O，阻塞IO），他的并发性能得到了很大提高，两张图让你了解BIO和NIO的区别：

阻塞IO的通信方式

非阻塞IO的通信方式

从这两图可以看出，NIO的单线程能处理连接的数量比BIO要高出很多，而为什么单线程能处理更多的连接呢？原因就是图二中出现的Selector。
当一个连接建立之后，他有两个步骤要做，第一步是接收完客户端发过来的全部数据，第二步是服务端处理完请求业务之后返回response给客户端。NIO和BIO的区别主要是在第一步。
在BIO中，等待客户端发数据这个过程是阻塞的，这样就造成了一个线程只能处理一个请求的情况，而机器能支持的最大线程数是有限的，这就是为什么BIO不能支持高并发的原因。
而NIO中，当一个Socket建立好之后，Thread并不会阻塞去接受这个Socket，而是将这个请求交给Selector，Selector会不断的去遍历所有的Socket，一旦有一个Socket建立完成，他会通知Thread，然后Thread处理完数据再返回给客户端——这个过程是不阻塞的，这样就能让一个Thread处理更多的请求了。

下面两张图是基于BIO的处理流程和netty的处理流程，辅助你理解两种方式的差别：

BIO的处理流程

NIO的处理流程

除了BIO和NIO之外，还有一些其他的IO模型，下面这张图就表示了五种IO模型的处理流程：

五种常见的IO模型：

BIO：同步阻塞IO，阻塞整个步骤，如果连接少，他的延迟是最低的，因为一个线程只处理一个连接，适用于少连接且延迟低的场景，比如说数据库连接。
NIO：同步非阻塞IO，阻塞业务处理但不阻塞数据接收，适用于高并发且处理简单的场景，比如聊天软件。
多路复用IO：他的两个步骤处理是分开的，也就是说，一个连接可能他的数据接收是线程a完成的，数据处理是线程b完成的，他比BIO能处理更多请求。
信号驱动IO：这种IO模型主要用在嵌入式开发，不参与讨论。
异步IO：他的数据请求和数据处理都是异步的，数据请求一次返回一次，适用于长连接的业务场景

以上摘自Linux IO模式及 select、poll、epoll详解

5. Netty为什么传输快

Netty的传输快其实也是依赖了NIO的一个特性——零拷贝。我们知道，Java的内存有堆内存、栈内存和字符串常量池等等，其中堆内存是占用内存空间最大的一块，也是Java对象存放的地方，一般我们的数据如果需要从IO读取到堆内存，中间需要经过Socket缓冲区，也就是说一个数据会被拷贝两次才能到达他的的终点，如果数据量大，就会造成不必要的资源浪费。
Netty针对这种情况，使用了NIO中的另一大特性——零拷贝，当他需要接收数据的时候，他会在堆内存之外开辟一块内存，数据就直接从IO读到了那块内存中去，在netty里面通过ByteBuf可以直接对这些数据进行直接操作，从而加快了传输速度。
下两图就介绍了两种拷贝方式的区别，摘自Linux 中的零拷贝技术，第 1 部分

传统数据拷贝

零拷贝

二、入门案例

1. 目标

开发一个简单的服务器端和客户端

客户端向服务器端发送 hello, world
服务器仅接收，不返回

2. 环境准备

导入netty依赖

<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.39.Final</version>
</dependency>

3. 服务端

@Slf4j
public class HelloServer {
    public static void main(String[] args) {
        //1. 启动器：负责组装Netty组件，启动服务器
        new ServerBootstrap()
                //2. 事件组：BossEventLoop、WorkerEventLoop(Selector, Thread)
                .group(new NioEventLoopGroup())
                //3. 服务器ServerSocketChannel的实现
                .channel(NioServerSocketChannel.class)
                //4. 自定义事件处理器：Boss负责处理连接，Worker(child)负责处理读写，决定了Worker(child)能够处理哪些事件(Handle)
                .childHandler(
                    //5. Channel：服务端与客户端之间读写的通道；Initializer：负责添加别的Handle并初始化
                    new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel nioSocketChannel) {
                        //6. 添加具体的Handle
                        nioSocketChannel.pipeline().addLast(new StringDecoder());  //ByteBuf转换为字符串
                        nioSocketChannel.pipeline().addLast(new ChannelInboundHandlerAdapter(){  //自定义Handle
                            /**
                             * 监听读取事件
                             * @param ctx
                             * @param msg 客户端发送的数据
                             */
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                log.debug("READ事件获取的信息：{}", msg);
                            }
                        });
                    }
                })
                .bind(8080);
    }
}

代码解读

负责组装Netty组件，启动服务器
创建 NioEventLoopGroup，可以简单理解为线程池 + Selector 后面会详细展开
选择服务 Scoket 实现类，其中 NioServerSocketChannel 表示基于 NIO 的服务器端实现，其它实现还有

为啥方法叫 childHandler，是接下来添加的处理器都是给 SocketChannel 用的，而不是给 ServerSocketChannel。ChannelInitializer 处理器（仅执行一次），它的作用是待客户端 SocketChannel 建立连接后，执行 initChannel 以便添加更多的处理器
SocketChannel 的处理器，解码 ByteBuf => String
SocketChannel 的业务处理器，使用上一个处理器的处理结果
ServerSocketChannel 绑定的监听端口

4. 客户端

public class HelloClient {
    public static void main(String[] args) throws InterruptedException {
        new Bootstrap()  //1
                .group(new NioEventLoopGroup())  //2
                .channel(NioSocketChannel.class)  //3
                .handler(new ChannelInitializer<NioSocketChannel>() {  //4
                    @Override
                    protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
                        nioSocketChannel.pipeline().addLast(new StringEncoder());  //9
                    }
                })
                .connect(new InetSocketAddress("localhost", 8080))  //5
                .sync()  //6
                .channel()  //7
                .writeAndFlush(new Date() + "Hello World");  //8
    }
}

代码解读

1 处，启动客户端
2 处，创建 NioEventLoopGroup，同 Server
3 处，选择客户 Socket 实现类，NioSocketChannel 表示基于 NIO 的客户端实现，其它实现还有

4 处，添加 SocketChannel 的处理器，ChannelInitializer 处理器（仅执行一次），它的作用是待客户端 SocketChannel 建立连接后，执行 initChannel 以便添加更多的处理器
5 处，指定要连接的服务器和端口
6 处，Netty 中很多方法都是异步的，如 connect，这时需要使用 sync 方法等待 connect 建立连接完毕
7 处，获取 channel 对象，它即为通道抽象，可以进行数据读写操作
8 处，写入消息并清空缓冲区
9 处，消息会经过通道 handler 处理，这里是将 String => ByteBuf 发出

数据经过网络传输，到达服务器端，服务器端 5 和 6 处的 handler 先后被触发，走完一个流程

注意：

不论客户端还是服务端 initChannel() 内的代码需要进行连接后才执行

5. 流程数量

6. 💡提示

一开始需要树立正确的观念

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

工序有多道(类似流水线)，合在一起就是 pipeline，pipeline 负责发布事件（读、读取完成…）传播给每个 handler， handler 对自己感兴趣的事件进行处理（重写了相应事件处理方法）
handler 分 Inbound 和 Outbound 两类

把 eventLoop 理解为处理数据的工人

工人可以管理多个 channel 的 io 操作，并且一旦工人负责了某个 channel，就要负责到底（绑定）
工人既可以执行 io 操作，也可以进行任务处理，每位工人有任务队列，队列里可以堆放多个 channel 的待处理任务，任务分为普通任务、定时任务
工人按照 pipeline 顺序，依次按照 handler 的规划（代码）处理数据，可以为每道工序指定不同的工人

三、组件

1. EventLoop

1. 概述

EventLoop：事件循环对象

EventLoop 本质是一个单线程执行器（同时维护了一个 Selector），里面有 run 方法处理 Channel 上源源不断的 io 事件，它的继承关系比较复杂

一条线是继承自 java.util.concurrent.ScheduledExecutorService 因此包含了线程池中所有的方法
另一条线是继承自 netty 自己的 io.netty.util.concurrent.OrderedEventExecutor，
- 提供了 boolean inEventLoop(Thread thread) 方法判断一个线程是否属于此 EventLoop
- 提供了 parent 方法来看看自己属于哪个 EventLoopGroup
  
  EventLoopGroup：事件循环组
  EventLoopGroup 是一组 EventLoop，Channel 一般会调用 EventLoopGroup 的 register 方法来绑定其中一个 EventLoop，后续这个 Channel 上的 io 事件都由此 EventLoop 来处理（保证了 io 事件处理时的线程安全）
继承自 netty 自己的 EventExecutorGroup
实现了 Iterable 接口提供遍历 EventLoop 的能力
另有 next 方法获取集合中下一个 EventLoop

NioEventLoopGroup：负责处理IO事件、普通任务、定时任务 DefaultEventLoopGroup：负责处理普通任务、定时任务

2. 案例演示EventLoop执行

@Test
public void test01() throws IOException {
    //参数指定线程数（不指定就是Cpu线程数的两倍）
    EventLoopGroup eventLoopGroup = new NioEventLoopGroup(2);
    System.out.println(eventLoopGroup.next());
    System.out.println(eventLoopGroup.next());
    System.out.println(eventLoopGroup.next());

    eventLoopGroup.next().submit(() -> {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        log.debug("ok1");
    });

    /*
     * param1：执行的线程
     * param2：执行的延迟时间
     * param3：执行的间隔时间
     * param4：时间单位
     * */
    eventLoopGroup.next().scheduleAtFixedRate(() -> {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        log.debug("ok2");
    }, 0, 1, TimeUnit.SECONDS);

    log.debug("main");

    System.in.read();
}

输出

io.netty.channel.DefaultEventLoop@60f82f98
io.netty.channel.DefaultEventLoop@35f983a6
io.netty.channel.DefaultEventLoop@60f82f98

也可以使用 for 循环

DefaultEventLoopGroup group = new DefaultEventLoopGroup(2);
for (EventExecutor eventLoop : group) {
    System.out.println(eventLoop);
}

输出

io.netty.channel.DefaultEventLoop@60f82f98
io.netty.channel.DefaultEventLoop@35f983a6

3 💡 优雅关闭

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

4. 演示 NioEventLoop 处理 io 事件

案例

服务端

@Slf4j
public class EventLoopServer {
    public static void main(String[] args) throws InterruptedException {
        new ServerBootstrap()
                .group(new NioEventLoopGroup())
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) {
                        ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) {
                                ByteBuf byteBuf = msg instanceof ByteBuf ? ((ByteBuf) msg) : null;
                                if (byteBuf != null) {
                                    log.debug(byteBuf.toString(Charset.defaultCharset()));
                                }
                            }
                        });
                    }
                }).bind(8080).sync();
    }
}

客户端

public class EventLoopClient {
    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 nioSocketChannel) throws Exception {
                        nioSocketChannel.pipeline().addLast(new StringEncoder());
                    }
                })
                .connect(new InetSocketAddress("localhost", 8080))
                .sync()
                .channel();

        //断点在此处运行 channel.writeAndFlush("aaa")
        System.out.println("");
    }
}

改进点一

服务器端两个 nio worker 工人

boss只负责ServerSocketChannel 上 accept事件；worker只负责socketChannel 上的读写

@Slf4j
public class EventLoopServer {
    public static void main(String[] args) throws InterruptedException {
        new ServerBootstrap()
                //细节一：boss只负责ServerSocketChannel 上 accept事件；worker只负责socketChannel 上的读写
                .group(new NioEventLoopGroup(1), new NioEventLoopGroup(2))
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) {
                        ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) {
                                ByteBuf byteBuf = msg instanceof ByteBuf ? ((ByteBuf) msg) : null;
                                if (byteBuf != null) {
                                    log.debug(byteBuf.toString(Charset.defaultCharset()));
                                }
                            }
                        });
                    }
                }).bind(8080).sync();
    }
}

客户端：启动三次，分别修改发送字符串为 zhangsan（第一次），lisi（第二次），wangwu（第三次）

public class EventLoopClient {
    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 nioSocketChannel) throws Exception {
                        nioSocketChannel.pipeline().addLast(new StringEncoder());
                    }
                })
                .connect(new InetSocketAddress("localhost", 8080))
                .sync()
                .channel();

        //断点在此处运行 channel.writeAndFlush("zhangsan")
        System.out.println("");
    }
}

打印结果

22:03:34 [DEBUG] [nioEventLoopGroup-3-1] c.i.o.EventLoopTest - zhangsan       
22:03:36 [DEBUG] [nioEventLoopGroup-3-1] c.i.o.EventLoopTest - zhangsan       
22:05:36 [DEBUG] [nioEventLoopGroup-3-2] c.i.o.EventLoopTest - lisi           
22:05:38 [DEBUG] [nioEventLoopGroup-3-2] c.i.o.EventLoopTest - lisi           
22:06:09 [DEBUG] [nioEventLoopGroup-3-1] c.i.o.EventLoopTest - wangwu        
22:06:11 [DEBUG] [nioEventLoopGroup-3-1] c.i.o.EventLoopTest - wangwu

可以看到两个工人轮流处理 channel，但工人与 channel 之间进行了绑定

改进点二

当某个Channel中处理的事件较慢时，一定会影响到同一个EventLoop中的其他线程的运行效率；可以多创建一个EventLoop帮助我们处理该次事件

再增加两个非 nio 工人

@Slf4j
public class EventLoopServer {
    public static void main(String[] args) throws InterruptedException {
        //添加两个非NIO工人(EventLoopGroup)
        DefaultEventLoopGroup defaultLoop = new DefaultEventLoopGroup(2);
        new ServerBootstrap()
                //细节一：boss只负责ServerSocketChannel 上 accept事件；worker只负责socketChannel 上的读写
                .group(new NioEventLoopGroup(1), new NioEventLoopGroup(2))
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) {
                        ch.pipeline().addLast("handle1", new ChannelInboundHandlerAdapter() {
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) {
                                ByteBuf byteBuf = msg instanceof ByteBuf ? ((ByteBuf) msg) : null;
                                if (byteBuf != null) {
                                    //byte[] buf = new byte[16];
                                    //ByteBuf len = byteBuf.readBytes(buf, 0, byteBuf.readableBytes());
                                    log.debug(byteBuf.toString(Charset.defaultCharset()));
                                    //让消息传递给下一个Handle
                                    ctx.fireChannelRead(msg);
                                }
                            }
                        });
                        /*
                        * param1：负责处理事件的EventLoopGroup
                        * param2：handle名称
                        * param3：自定义处理事件Handle
                        * */
                        ch.pipeline().addLast(defaultLoop, "handle2", new ChannelInboundHandlerAdapter() {
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) {
                                ByteBuf byteBuf = msg instanceof ByteBuf ? ((ByteBuf) msg) : null;
                                if (byteBuf != null) {
                                    //byte[] buf = new byte[16];
                                    //ByteBuf len = byteBuf.readBytes(buf, 0, byteBuf.readableBytes());
                                    log.debug(byteBuf.toString(Charset.defaultCharset()));
                                }
                            }
                        });
                    }
                }).bind(8080).sync();
    }
}

客户端不变，启动三次，分别修改发送字符串为 zhangsan（第一次），lisi（第二次），wangwu（第三次）

打印结果

23:40:50 [DEBUG] [nioEventLoopGroup-4-1] c.g.n.EventLoopServer - wangwu
23:40:50 [DEBUG] [defaultEventLoopGroup-2-1] c.g.n.EventLoopServer - wangwu
23:40:53 [DEBUG] [nioEventLoopGroup-4-2] c.g.n.EventLoopServer - zhangsan
23:40:53 [DEBUG] [defaultEventLoopGroup-2-2] c.g.n.EventLoopServer - zhangsan
23:40:54 [DEBUG] [nioEventLoopGroup-4-1] c.g.n.EventLoopServer - lisi
23:40:54 [DEBUG] [defaultEventLoopGroup-2-1] c.g.n.EventLoopServer - lisi

可以看到，nio 工人和非 nio 工人也分别绑定了 channel（LoggingHandler 由 nio 工人执行，而我们自己的 handler 由非 nio 工人执行）

5. 💡 handler 执行中如何换人？

关键代码 _io.netty.channel.AbstractChannelHandlerContext#invokeChannelRead()_

如果两个 handler 绑定的是同一个线程，那么就直接调用
否则，把要调用的代码封装为一个任务对象，由下一个 handler 的线程来调用

2. Channel

1. 常用方法

channel 的主要作用

close() 可以用来关闭 channel
closeFuture() 用来处理 channel 的关闭
- sync 方法作用是同步等待 channel 关闭
- 而 addListener 方法是异步等待 channel 关闭
pipeline() 方法添加处理器
write() 方法将数据写入
writeAndFlush() 方法将数据写入并刷出

2. ChannelFuture

这是刚刚的的客户端代码

new Bootstrap()
    .group(new NioEventLoopGroup())
    .channel(NioSocketChannel.class)
    .handler(new ChannelInitializer<NioSocketChannel>() {
        @Override
        protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
            nioSocketChannel.pipeline().addLast(new StringEncoder());
        }
    })
    .connect(new InetSocketAddress("localhost", 8080))
    .sync()
    .channel();

现在将它拆开来看

ChannelFuture channelFuture = new Bootstrap()
                .group(new NioEventLoopGroup())
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
                        nioSocketChannel.pipeline().addLast(new StringEncoder());
                    }
                })
                //connect() 返回的是 ChannelFuture 对象，它的作用是利用 channel() 方法来获取 Channel 对象
                //异步非阻塞方法，main线程发起调用，建立连接的是NIO线程
                .connect(new InetSocketAddress("localhost", 8080));

        channelFuture.sync();
        Channel channel = channelFuture.channel();
        channel.writeAndFlush("hello world");
        log.debug("{}", channel);

注意： connect 方法是异步的，意味着不等连接建立，方法执行就返回了。因此 channelFuture 对象中不能【立刻】获得到正确的 Channel 对象补充：带有 Future、Promise 的类型都是和异步方法搭配使用，用来处理结果

实验如下：

ChannelFuture channelFuture = new Bootstrap()
    .group(new NioEventLoopGroup())
    .channel(NioSocketChannel.class)
    .handler(new ChannelInitializer<NioSocketChannel>() {
        @Override
        protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
            nioSocketChannel.pipeline().addLast(new StringEncoder());
        }
    })
    //1. connect()连接到服务器
    //异步非阻塞方法，main线程发起调用，建立连接的是NIO线程
    .connect(new InetSocketAddress("localhost", 8080));

//使用sync方法同步处理结果，阻塞当前线程，直到NIO线程建立连接完毕
//channelFuture.sync();  //注释该同步代码
Channel channel = channelFuture.channel();
channel.writeAndFlush("hello world");
log.debug("{}", channel);

连接未建立，打印 [id: 0x4f009297]
查看服务器端未接收到客户端响应的数据

除了用 sync 方法可以让异步操作同步以外，还可以使用回调的方式：

//使用 addListener(回调对象) 方法异步处理结果
channelFuture.addListener(new ChannelFutureListener() {
    /**
     * @param channelFuture
     * @description 在NIO线程将连接建立好之后调用
     */
    @Override
    public void operationComplete(ChannelFuture channelFuture) throws Exception {
        Channel channel = channelFuture.channel();
        log.debug("{}", channel);
        channel.writeAndFlush("hello world");
    }
});

打印结果：
客户端

01:36:15 [DEBUG] [nioEventLoopGroup-2-1] c.g.n.EventLoopClient - [id: 0x90786e4a, L:/127.0.0.1:7384 - R:localhost/127.0.0.1:8080]

发现与服务端已经成功建立了连接，并且是在NIO线程内打印的数据

服务端

01:36:15 [DEBUG] [nioEventLoopGroup-4-2] c.g.n.EventLoopServer - hello world
01:36:15 [DEBUG] [defaultEventLoopGroup-2-2] c.g.n.EventLoopServer - hello world

3. CloseFuture

@Slf4j
public class CloseFutureClient {
    public static void main(String[] args) throws InterruptedException {
        //取消与服务端的连接需要关闭EventLoopGroup里的线程
        NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        ChannelFuture channelFuture = new Bootstrap()
                .group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
                        nioSocketChannel.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
                        nioSocketChannel.pipeline().addLast(new StringEncoder());
                    }
                })
                //1. connect()连接到服务器
                //异步非阻塞方法，main线程发起调用，建立连接的是NIO线程
                .connect(new InetSocketAddress("localhost", 8080));

        //使用 addListener(回调对象) 方法异步处理结果
        channelFuture.addListener(new ChannelFutureListener() {
            /**
             * @param channelFuture
             * @description 在NIO线程将连接建立好之后调用
             */
            @Override
            public void operationComplete(ChannelFuture channelFuture) throws Exception {
                Channel channel = channelFuture.channel();

                new Thread(() -> {
                    while (true){
                        System.out.print("请输入：");
                        String input = new Scanner(System.in).next();
                        if ("q".equals(input)){
                            //channel.close()是一个异步方法
                            channel.close();
                            //在这里进行关闭之后操作，可能会导致线程执行先后问题
                            //log.debug("处理关闭之后的操作");
                            break;
                        }
                        channel.writeAndFlush(input);
                    }
                }, "input").start();

                // 获取 CloseFuture 对象
                ChannelFuture closeFuture = channel.closeFuture();

                //1.同步处理关闭（存在问题）
//                log.debug("waiting close...");
//                closeFuture.sync();
//                log.debug("处理关闭之后的操作");
//                eventLoopGroup.shutdownGracefully();

                //2.异步处理关闭
                closeFuture.addListener((ChannelFutureListener) channelFuture2 -> {
                    log.debug("处理关闭之后的操作");
                    eventLoopGroup.shutdownGracefully();
                });
            }
        });
    }
}

channel.close()：关闭与服务端的连接（该方法是异步执行所以往后执行的代码会存在问题）
channel.closeFuture()：获取一个监听关闭事件的ChannelFuture（可使用其进行同步阻塞/异步非阻塞的结果处理）

💡补充：

关闭与服务端的连接后，由于NioEventLoopGroup中任然存在等待的线程所以客户端不会关闭使用 shutdownGracefully() 方法温和地关闭内部地线程

4. 💡 异步提升的是什么

看到这里会有疑问：为什么不在一个线程中去执行建立连接、去执行关闭 channel，那样不是也可以吗？非要用这么复杂的异步方式：比如一个线程发起建立连接，另一个线程去真正建立连接
还有同学会笼统地回答，因为 netty 异步方式用了多线程、多线程就效率高。其实这些认识都比较片面，多线程和异步所提升的效率并不是所认为的

思考下面的场景，4 个医生给人看病，每个病人花费 20 分钟，而且医生看病的过程中是以病人为单位的，一个病人看完了，才能看下一个病人。假设病人源源不断地来，可以计算一下 4 个医生一天工作 8 小时，处理的病人总数是：4 * 8 * 3 = 96

经研究发现，看病可以细分为四个步骤，经拆分后每个步骤需要 5 分钟，如下

原本一个医生一小时只能处理 3 个病人，任务拆分后一小时可以处理十二个病人。这提升了单位时间的处理任务的吞吐量

要点

单线程没法异步提高效率，必须配合多线程、多核 cpu 才能发挥异步的优势
异步并没有缩短响应时间，反而有所增加
合理进行任务拆分，也是利用异步的关键

3. Future & Promise

1. 概述

在异步处理时，经常用到这两个接口
首先要说明 netty 中的 Future 与 jdk 中的 Future 同名，但是是两个接口，netty 的 Future 继承自 jdk 的 Future，而 Promise 又对 netty Future 进行了扩展

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

功能/名称	jdk Future	netty Future	Promise
cancel	取消任务	-	-
isCanceled	任务是否取消	-	-
isDone	任务是否完成，不能区分成功失败	-	-
get	获取任务结果，阻塞等待	-	-
getNow	-	获取任务结果，非阻塞，还未产生结果时返回 null	-
await	-	等待任务结束，如果任务失败，不会抛异常，而是通过 isSuccess 判断	-
sync	-	等待任务结束，如果任务失败，抛出异常	-
isSuccess	-	判断任务是否成功	-
cause	-	获取失败信息，非阻塞，如果没有失败，返回null	-
addLinstener	-	添加回调，异步接收结果	-
setSuccess	-	-	设置成功结果
setFailure	-	-	设置失败结果

2. JDK Future

使用案例

@Slf4j
public class FutureAndPromise {
    static void jdkFuture() throws ExecutionException, InterruptedException {
        //创建线程池
        ExecutorService threadPool = Executors.newCachedThreadPool();
        //提交线程任务
        Future<?> future = threadPool.submit((Callable<Integer>) () -> {
            try {
                log.debug("开始计算结果");
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return 40;
        });

        log.debug("等待计算结果");
        //future.get()：阻塞等待结果
        log.debug("计算结果={}", future.get());

        threadPool.shutdown();
    }
}

打印结果

14:15:00 [DEBUG] [main] c.g.n.FutureAndPromise - 等待计算结果
14:15:00 [DEBUG] [pool-1-thread-1] c.g.n.FutureAndPromise - 开始计算结果
14:15:01 [DEBUG] [main] c.g.n.FutureAndPromise - 计算结果=40

仍然是由main线程处理返回结果，所以这里使用的是同步

3. Netty Future

@Slf4j
public class FutureAndPromise {
    static void nettyFuture(){
        //创建线程组
        DefaultEventLoopGroup eventLoopGroup = new DefaultEventLoopGroup(1);

        //提交线程任务
        io.netty.util.concurrent.Future<Integer> future = eventLoopGroup.submit(() -> {
            try {
                log.debug("开始计算结果");
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return 40;
        });

        //使用Netty Future异步处理返回结果
        future.addListener(new GenericFutureListener<io.netty.util.concurrent.Future<? super Integer>>() {
            @Override
            public void operationComplete(io.netty.util.concurrent.Future<? super Integer> future) throws Exception {
                log.debug("得出计算结果={}", future.getNow());
            }
        });

        eventLoopGroup.shutdownGracefully();
    }
}

打印计算结果

15:24:37 [DEBUG] [defaultEventLoopGroup-2-1] c.g.n.FutureAndPromise - 开始计算结果
15:24:38 [DEBUG] [defaultEventLoopGroup-2-1] c.g.n.FutureAndPromise - 得出计算结果=40

使用的是defaultEventLoopGroup线程处理返回结果，所以这里使用的是异步

4. Netty Promise

案例一：同步处理任务

@Test
public void nettyPromise1() throws ExecutionException, InterruptedException {
    DefaultEventLoop eventExecutors = new DefaultEventLoop();
    DefaultPromise<Integer> promise = new DefaultPromise<>(eventExecutors);

    eventExecutors.execute(()->{
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        log.debug("set success, {}",10);
        promise.setSuccess(10);
    });

    log.debug("start...");
    log.debug("{}",promise.getNow()); // 还没有结果
    log.debug("{}",promise.get());
}

打印结果

15:27:05 [DEBUG] [main] c.g.n.FutureAndPromise - start...
15:27:05 [DEBUG] [main] c.g.n.FutureAndPromise - null
15:27:06 [DEBUG] [defaultEventLoop-1-1] c.g.n.FutureAndPromise - set success, 10
15:27:06 [DEBUG] [main] c.g.n.FutureAndPromise - 10

案例二：异步处理任务

@Test
public void nettyPromise2() throws IOException {
    DefaultEventLoop eventExecutors = new DefaultEventLoop();
    DefaultPromise<Integer> promise = new DefaultPromise<>(eventExecutors);

    // 设置回调，异步接收结果
    promise.addListener(future -> {
        // 这里的 future 就是上面的 promise
        log.debug("{}",future.getNow());
    });

    // 等待 1000 后设置成功结果
    eventExecutors.execute(()->{
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        log.debug("set success, {}",10);
        promise.setSuccess(10);
    });

    log.debug("start...");

    System.in.read();
}

打印结果

15:53:35 [DEBUG] [main] c.g.n.FutureAndPromise - start...
15:53:36 [DEBUG] [defaultEventLoop-1-1] c.g.n.FutureAndPromise - set success, 10
15:53:36 [DEBUG] [defaultEventLoop-1-1] c.g.n.FutureAndPromise - 10

案例三：同步处理任务失败

Sync

@Test
public void nettyPromise3() throws IOException, ExecutionException, InterruptedException {
    DefaultEventLoop eventExecutors = new DefaultEventLoop();
    DefaultPromise<Integer> promise = new DefaultPromise<>(eventExecutors);

    eventExecutors.execute(() -> {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        RuntimeException e = new RuntimeException("error...");
        log.debug("set failure, {}", e.toString());
        promise.setFailure(e);
    });

    log.debug("start...");
    log.debug("{}", promise.getNow());
    promise.get(); // sync() 也会出现异常，只是 get 会再用 ExecutionException 包一层异常

    System.in.read();
}

打印结果

Await

@Test
public void nettyPromise4() throws InterruptedException, IOException {
    DefaultEventLoop eventExecutors = new DefaultEventLoop();
    DefaultPromise<Integer> promise = new DefaultPromise<>(eventExecutors);

    eventExecutors.execute(() -> {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        RuntimeException e = new RuntimeException("error...");
        log.debug("set failure, {}", e.toString());
        promise.setFailure(e);
    });

    log.debug("start...");
    log.debug("{}", promise.getNow());
    promise.await(); // 与 sync 和 get 区别在于，不会抛异常
    log.debug("result {}", (promise.isSuccess() ? promise.getNow() : promise.cause()).toString());

    System.in.read();
}

打印结果

案例四：异步处理任务失败

@Test
public void nettyPromise5() throws IOException {
    DefaultEventLoop eventExecutors = new DefaultEventLoop();
    DefaultPromise<Integer> promise = new DefaultPromise<>(eventExecutors);

    promise.addListener(future -> {
        log.debug("result {}", (promise.isSuccess() ? promise.getNow() : promise.cause()).toString());
    });

    eventExecutors.execute(() -> {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        RuntimeException e = new RuntimeException("error...");
        log.debug("set failure, {}", e.toString());
        promise.setFailure(e);
    });

    log.debug("start...");

    System.in.read();
}

打印结果

案例五：Await死锁

DefaultEventLoop eventExecutors = new DefaultEventLoop();
DefaultPromise<Integer> promise = new DefaultPromise<>(eventExecutors);

eventExecutors.submit(()->{
    System.out.println("1");
    try {
        promise.await();
        // 注意不能仅捕获 InterruptedException 异常
        // 否则 死锁检查抛出的 BlockingOperationException 会继续向上传播
        // 而提交的任务会被包装为 PromiseTask，它的 run 方法中会 catch 所有异常然后设置为 Promise 的失败结果而不会抛出
    } catch (Exception e) { 
        e.printStackTrace();
    }
    System.out.println("2");
});
eventExecutors.submit(()->{
    System.out.println("3");
    try {
        promise.await();
    } catch (Exception e) {
        e.printStackTrace();
    }
    System.out.println("4");
});

打印结果

1
2
3
4
io.netty.util.concurrent.BlockingOperationException: DefaultPromise@47499c2a(incomplete)
    at io.netty.util.concurrent.DefaultPromise.checkDeadLock(DefaultPromise.java:384)
    at io.netty.util.concurrent.DefaultPromise.await(DefaultPromise.java:212)
    at com.itcast.oio.DefaultPromiseTest.lambda$main$0(DefaultPromiseTest.java:27)
    at io.netty.util.concurrent.PromiseTask$RunnableAdapter.call(PromiseTask.java:38)
    at io.netty.util.concurrent.PromiseTask.run(PromiseTask.java:73)
    at io.netty.channel.DefaultEventLoop.run(DefaultEventLoop.java:54)
    at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:918)
    at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
    at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
    at java.lang.Thread.run(Thread.java:745)
io.netty.util.concurrent.BlockingOperationException: DefaultPromise@47499c2a(incomplete)
    at io.netty.util.concurrent.DefaultPromise.checkDeadLock(DefaultPromise.java:384)
    at io.netty.util.concurrent.DefaultPromise.await(DefaultPromise.java:212)
    at com.itcast.oio.DefaultPromiseTest.lambda$main$1(DefaultPromiseTest.java:36)
    at io.netty.util.concurrent.PromiseTask$RunnableAdapter.call(PromiseTask.java:38)
    at io.netty.util.concurrent.PromiseTask.run(PromiseTask.java:73)
    at io.netty.channel.DefaultEventLoop.run(DefaultEventLoop.java:54)
    at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:918)
    at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
    at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
    at java.lang.Thread.run(Thread.java:745)

5. Handler & Pipeline

ChannelHandler 用来处理 Channel 上的各种事件，分为入站、出站两种。所有 ChannelHandler 被连成一串，就是 Pipeline

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

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

编写服务端代码，查看执行顺序

@Slf4j
public class PipelineServer {
    public static void main(String[] args) {
        new ServerBootstrap()
                .group(new NioEventLoopGroup())
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
                        ChannelPipeline pipeline = nioSocketChannel.pipeline();

                        pipeline.addLast("handler1", new ChannelInboundHandlerAdapter(){
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                ByteBuf byteBuf = (ByteBuf) msg;
                                String name = byteBuf.toString(Charset.defaultCharset());
                                log.debug("handler1处理结果{}", name);
                                super.channelRead(ctx, name);
                            }
                        });

                        pipeline.addLast("handler2", new ChannelInboundHandlerAdapter(){
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                Student student = new Student(msg.toString());
                                log.debug("handler2处理结果{}", student);
                                super.channelRead(ctx, student);
                            }
                        });

                        pipeline.addLast("handler3", new ChannelInboundHandlerAdapter(){
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                log.debug("handler3处理结果{}", msg.getClass());
                                //super.channelRead(ctx, msg);
                                ctx.channel().writeAndFlush(msg);
                            }
                        });

                        pipeline.addLast("outHandler1" ,new ChannelOutboundHandlerAdapter(){
                            @Override
                            public void write(ChannelHandlerContext ctx, Object msg,
                                              ChannelPromise promise) {
                                log.debug("outHandler1接收到数据{}", msg);
                                ctx.write(msg, promise); // 4
                            }
                        });
                        pipeline.addLast("outHandler2" ,new ChannelOutboundHandlerAdapter(){
                            @Override
                            public void write(ChannelHandlerContext ctx, Object msg,
                                              ChannelPromise promise) {
                                log.debug("outHandler2接收到数据{}", msg);
                                ctx.write(msg, promise); // 5
                            }
                        });
                        pipeline.addLast("outHandler3" ,new ChannelOutboundHandlerAdapter(){
                            @Override
                            public void write(ChannelHandlerContext ctx, Object msg,
                                              ChannelPromise promise) {
                                log.debug("outHandler3接收到数据{}", msg);
                                ctx.write(msg, promise); // 6
                            }
                        });
                    }
                })
                .bind(new InetSocketAddress(8080));
    }

    static class Student{
        String name;

        public Student(String name) {
            this.name = name;
        }

        @Override
        public String toString() {
            return "Student{" +
                    "name='" + name + '\'' +
                    '}';
        }
    }
}

客户端

@Slf4j
public class CloseFutureClient {
    public static void main(String[] args) throws InterruptedException {
        NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        ChannelFuture channelFuture = new Bootstrap()
                .group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
                        nioSocketChannel.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
                        nioSocketChannel.pipeline().addLast(new StringEncoder());
                    }
                })
                .connect(new InetSocketAddress("localhost", 8080));

        //使用 addListener(回调对象) 方法异步处理结果
        channelFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture channelFuture) throws Exception {
                Channel channel = channelFuture.channel();

                new Thread(() -> {
                    while (true){
                        System.out.print("请输入：");
                        String input = new Scanner(System.in).next();
                        if ("q".equals(input)){
                            channel.close();
                            break;
                        }
                        channel.writeAndFlush(input);
                    }
                }, "input").start();

                // 获取 CloseFuture 对象
                ChannelFuture closeFuture = channel.closeFuture();

                // 异步处理关闭
                closeFuture.addListener((ChannelFutureListener) channelFuture2 -> {
                    log.debug("处理关闭之后的操作");
                    eventLoopGroup.shutdownGracefully();
                });
            }
        });
    }
}

打印结果

17:06:23 [DEBUG] [nioEventLoopGroup-2-2] c.g.n.PipelineServer - handler1处理结果qwe
17:06:23 [DEBUG] [nioEventLoopGroup-2-2] c.g.n.PipelineServer - handler2处理结果Student{name='qwe'}
17:06:23 [DEBUG] [nioEventLoopGroup-2-2] c.g.n.PipelineServer - handler3处理结果class com.gmf.netty.PipelineServer$Student
17:06:23 [DEBUG] [nioEventLoopGroup-2-2] c.g.n.PipelineServer - outHandler3接收到数据Student{name='qwe'}
17:06:23 [DEBUG] [nioEventLoopGroup-2-2] c.g.n.PipelineServer - outHandler2接收到数据Student{name='qwe'}
17:06:23 [DEBUG] [nioEventLoopGroup-2-2] c.g.n.PipelineServer - outHandler1接收到数据Student{name='qwe'}

可以看到，ChannelInboundHandlerAdapter 是按照 addLast 的顺序执行的，而 ChannelOutboundHandlerAdapter 是按照 addLast 的逆序执行的。ChannelPipeline 的实现是一个 ChannelHandlerContext（包装了 ChannelHandler）组成的双向链表

每次addLast()都是向头尾节点中间添加Handler，这两个节点是Netty中自带的

super.channelRead(ctx, name) 内部调用 ctx.fireChannelRead(msg)
ctx.fireChannelRead(msg) 是 调用下一个入站处理器
ctx.channel().write(msg) 会 从调用链尾部开始触发 后续出站处理器的执行
ctx.write(msg, promise) 的调用也会 触发上一个出站处理器

ctx.channel().write(msg) VS ctx.write(msg)

都是触发出站处理器的执行
ctx.channel().write(msg) 从尾部开始查找出站处理器
ctx.write(msg) 是从当前节点找上一个出站处理器
handler3 执行的 ctx.channel().write(msg) 如果改为 ctx.write(msg) 仅会打印 1 2 3，因为节点3 之前没有其它出站处理器了
outHandler3 处的 ctx.write(msg, promise) 如果改为 ctx.channel().write(msg) 会打印 1 2 3 6 6 6… 因为 ctx.channel().write() 是从尾部开始查找，结果又是节点6 自己

服务端 pipeline 触发的原始流程，图中数字代表了处理步骤的先后次序

6. EmbeddedChannel基本使用

@Slf4j
public class TestEmbeddedChannel {
    public static void main(String[] args) {
        ChannelInboundHandlerAdapter h1 = new ChannelInboundHandlerAdapter() {
            @Override
            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                log.debug("inHandler1");
                super.channelRead(ctx, msg);
            }
        };

        ChannelInboundHandlerAdapter h2 = new ChannelInboundHandlerAdapter() {
            @Override
            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                log.debug("inHandler2");
                super.channelRead(ctx, msg);
            }
        };

        ChannelOutboundHandlerAdapter h3 = new ChannelOutboundHandlerAdapter() {
            @Override
            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
                log.debug("outHandler1");
                super.write(ctx, msg, promise);
            }
        };

        ChannelOutboundHandlerAdapter h4 = new ChannelOutboundHandlerAdapter() {
            @Override
            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
                log.debug("outHandler2");
                super.write(ctx, msg, promise);
            }
        };

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

        //17:37:37 [DEBUG] [main] c.g.n.TestEmbeddedChannel - inHandler1
        //17:37:37 [DEBUG] [main] c.g.n.TestEmbeddedChannel - inHandler2
        embeddedChannel.writeInbound(ByteBufAllocator.DEFAULT.buffer().writeBytes("InBound".getBytes(StandardCharsets.UTF_8)));

        //17:38:37 [DEBUG] [main] c.g.n.TestEmbeddedChannel - outHandler2
        //17:38:37 [DEBUG] [main] c.g.n.TestEmbeddedChannel - outHandler1
        embeddedChannel.writeOutbound(ByteBufAllocator.DEFAULT.buffer().writeBytes("OutBound".getBytes(StandardCharsets.UTF_8)));
    }
}

四、ByteBuf

ByteBuf：是对字节数据的封装

1. 创建

@Test
public void create(){
    ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer();
    log(buffer);

    StringBuilder s = new StringBuilder();
    for (int i = 0; i < 300; i++) {
        s.append("a");
    }

    buffer.writeBytes(s.toString().getBytes(StandardCharsets.UTF_8));
    log(buffer);
}

上面代码创建了一个默认的 ByteBuf（池化基于直接内存的 ByteBuf），初始容量是 256

执行结果

其中 log() 方法参考

private 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);  //import static io.netty.util.internal.StringUtil.NEWLINE;
    appendPrettyHexDump(buf, buffer);  //import static io.netty.buffer.ByteBufUtil.appendPrettyHexDump;
    System.out.println(buf);
}

2. 直接内存 vs 堆内存

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

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

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

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

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

3. 池化 vs 非池化

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

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

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

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

4.1 以后，非 Android 平台默认启用池化实现，Android 平台启用非池化实现
4.1 之前，池化功能还不成熟，默认是非池化实现

配置IDEA虚拟环境变量

再次查看ByteBuf对象

4. ByteBuf的组成

ByteBuf 由四部分组成

最开始读写指针都在0位置

与ByteBuffer相比，读写指针进行了分离，这样就不需要在每次使用前切换读写状态

5. 写入

方法列表，省略一些不重要的方法

方法签名	含义	备注
writeBoolean(boolean value)	写入 boolean 值	用一字节 01\|00 代表 true\|false
writeByte(int value)	写入 byte 值
writeShort(int value)	写入 short 值
writeInt(int value)	写入 int 值	Big Endian，即 0x250，写入后 00 00 02 50
writeIntLE(int value)	写入 int 值	Little Endian，即 0x250，写入后 50 02 00 00
writeLong(long value)	写入 long 值
writeChar(int value)	写入 char 值
writeFloat(float value)	写入 float 值
writeDouble(double value)	写入 double 值
writeBytes(ByteBuf src)	写入 netty 的 ByteBuf
writeBytes(byte[] src)	写入 byte[]
writeBytes(ByteBuffer src)	写入 nio 的 ByteBuffer
int writeCharSequence(CharSequence sequence, Charset charset)	写入字符串

注意

这些方法的未指明返回值的，其返回值都是 ByteBuf，意味着可以链式调用
网络传输，默认习惯是 Big Endian

先写入 4 个字节

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

结果是

 read index:0 write index:4 capacity:10
          +-------------------------------------------------+
          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
 +--------+-------------------------------------------------+----------------+
 |00000000| 01 02 03 04                                     |....            |
 +--------+-------------------------------------------------+----------------+

再写入一个 int 整数，也是 4 个字节

buffer.writeInt(5);
log(buffer);

结果是

 read index:0 write index:8 capacity:10
          +-------------------------------------------------+
          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
 +--------+-------------------------------------------------+----------------+
 |00000000| 01 02 03 04 00 00 00 05                         |........        |
 +--------+-------------------------------------------------+----------------+

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

6. 扩容

再写入一个 int 整数时，容量不够了（初始容量是 10），这时会引发扩容

buffer.writeInt(6);
log(buffer);

扩容规则是

如何写入后数据大小未超过 512，则选择下一个 16 的整数倍，例如写入后大小为 12 ，则扩容后 capacity 是 16
如果写入后数据大小超过 512，则选择下一个 2^n，例如写入后大小为 513，则扩容后 capacity 是 2^10=1024（2^9=512 已经不够了）
扩容不能超过 max capacity 会报错

结果是

 read index:0 write index:12 capacity:16
          +-------------------------------------------------+
          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
 +--------+-------------------------------------------------+----------------+
 |00000000| 01 02 03 04 00 00 00 05 00 00 00 06             |............    |
 +--------+-------------------------------------------------+----------------+

7. 读取

假设原有12个字符

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| 31 32 33 34 35 36 37 38 39 31 30 31             |123456789101    |
+--------+-------------------------------------------------+----------------+

例如读了 4 次，每次一个字节

buffer.readByte();
buffer.readByte();
buffer.readByte();
buffer.readByte();
log(buffer);

如果需要重复读取 int 整数 5，怎么办？可以在 read 前先做个标记 mark

buffer.markReaderIndex();
System.out.println(buffer.readInt());
log(buffer);

打印结果

5
read index:8 write index:12 capacity:16
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 00 00 00 06                                     |....            |
+--------+-------------------------------------------------+----------------+

这时要重复读取的话，重置到标记位置 reset

buffer.resetReaderIndex();
log(buffer);

打印结果

read index:4 write index:12 capacity:16
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 00 00 00 05 00 00 00 06                         |........        |
+--------+-------------------------------------------------+----------------+

还有种办法是采用 get 开头的一系列方法，这些方法不会改变 read index

8. retain & release

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

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

回收内存的源码实现，请关注下面方法的不同实现 _protected abstract void deallocate()_

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

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

谁来负责 release 呢？
不是我们想象的（一般情况下）

 ByteBuf buf = ...
 try {
     ...
 } finally {
     buf.release();
 }

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

基本规则是，谁是最后使用者，谁负责 release，详细分析如下

起点，对于 NIO 实现来讲，在 io.netty.channel.nio.AbstractNioByteChannel.NioByteUnsafe#read 方法中首次创建 ByteBuf 放入 pipeline（line 163 pipeline.fireChannelRead(byteBuf)）
入站 ByteBuf 处理原则
- 对原始 ByteBuf 不做处理，调用 ctx.fireChannelRead(msg) 向后传递，这时无须 release
- 将原始 ByteBuf 转换为其它类型的 Java 对象，这时 ByteBuf 就没用了，必须 release
- 如果不调用 ctx.fireChannelRead(msg) 向后传递，那么也必须 release
- 注意各种异常，如果 ByteBuf 没有成功传递到下一个 ChannelHandler，必须 release
- 假设消息一直向后传，那么 TailContext 会负责释放未处理消息（原始的 ByteBuf）
出站 ByteBuf 处理原则
- 出站消息最终都会转为 ByteBuf 输出，一直向前传，由 HeadContext flush 后 release
异常处理原则
- 有时候不清楚 ByteBuf 被引用了多少次，但又必须彻底释放，可以循环调用 release 直到返回 true

TailContext 释放未处理消息逻辑

// io.netty.channel.DefaultChannelPipeline#onUnhandledInboundMessage(java.lang.Object)
protected void onUnhandledInboundMessage(Object msg) {
    try {
        logger.debug(
            "Discarded inbound message {} that reached at the tail of the pipeline. " +
            "Please check your pipeline configuration.", msg);
    } finally {
        ReferenceCountUtil.release(msg);
    }
}

HeadContext 释放未处理消息逻辑

//io.netty.channel.DefaultChannelPipeline.HeadContext#write 中调用 io.netty.channel.AbstractChannel.AbstractUnsafe#write
public final void write(Object msg, ChannelPromise promise) {
    this.assertEventLoop();
    ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
    if (outboundBuffer == null) {
        this.safeSetFailure(promise, this.newClosedChannelException(AbstractChannel.this.initialCloseCause));
        ReferenceCountUtil.release(msg);
    } else {
        int size;
        try {
            msg = AbstractChannel.this.filterOutboundMessage(msg);
            size = AbstractChannel.this.pipeline.estimatorHandle().size(msg);
            if (size < 0) {
                size = 0;
            }
        } catch (Throwable var6) {
            this.safeSetFailure(promise, var6);
            ReferenceCountUtil.release(msg);
            return;
        }

        outboundBuffer.addMessage(msg, size, promise);
    }
}

ReferenceCountUtil.release(msg)

// io.netty.util.ReferenceCountUtil#release(java.lang.Object)
public static boolean release(Object msg) {
    if (msg instanceof ReferenceCounted) {
        return ((ReferenceCounted) msg).release();
    }
    return false;
}

9. ByteBuf零拷贝

1. Slice

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

例，原始 ByteBuf 进行一些初始操作

ByteBuf origin = ByteBufAllocator.DEFAULT.buffer(10);
origin.writeBytes(new byte[]{1, 2, 3, 4});
origin.readByte();
System.out.println(ByteBufUtil.prettyHexDump(origin));

输出

         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 02 03 04                                        |...             |
+--------+-------------------------------------------------+----------------+

这时调用 slice 进行切片，无参 slice 是从原始 ByteBuf 的 read index 到 write index 之间的内容进行切片，切片后的 max capacity 被固定为这个区间的大小，因此不能追加 write

ByteBuf slice = origin.slice();
System.out.println(ByteBufUtil.prettyHexDump(slice));
// slice.writeByte(5); 如果执行，会报 IndexOutOfBoundsException 异常

输出

         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 02 03 04                                        |...             |
+--------+-------------------------------------------------+----------------+

如果原始 ByteBuf 再次读操作（又读了一个字节）

origin.readByte();
System.out.println(ByteBufUtil.prettyHexDump(origin));

输出

         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 03 04                                           |..              |
+--------+-------------------------------------------------+----------------+

这时的 slice 不受影响，因为它有独立的读写指针

System.out.println(ByteBufUtil.prettyHexDump(slice));

输出

         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 02 03 04                                        |...             |
+--------+-------------------------------------------------+----------------+

如果 slice 的内容发生了更改

slice.setByte(1, 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 05 04                                        |...             |
+--------+-------------------------------------------------+----------------+

这时，原始 ByteBuf 也会受影响，因为底层都是同一块内存

System.out.println(ByteBufUtil.prettyHexDump(origin));

输出

         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 05 04                                           |..              |
+--------+-------------------------------------------------+----------------+

slice(int index, int length) ：将ByteBuf从指定索引处切割指定长度的ByteBuf

ByteBuf origin = ByteBufAllocator.DEFAULT.buffer(10);
origin.writeBytes(new byte[]{0, 1, 2, 3, 4, 5, 6, 7, 8, 9});
System.out.println(ByteBufUtil.prettyHexDump(origin));

//使用slice()方法对ByteBuf进行切片
ByteBuf slice1 = origin.slice(0, 5);
ByteBuf slice2 = origin.slice(5, 5);
System.out.println(ByteBufUtil.prettyHexDump(slice1));
System.out.println(ByteBufUtil.prettyHexDump(slice2));

查看输出

         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 00 01 02 03 04 05 06 07 08 09                   |..........      |
+--------+-------------------------------------------------+----------------+
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 00 01 02 03 04                                  |.....           |
+--------+-------------------------------------------------+----------------+
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 05 06 07 08 09                                  |.....           |
+--------+-------------------------------------------------+----------------+

总结： slice() 方法切割的对象，只是在原有的ByteBuf的上创建出独属的读写指针，并且与原ByteBuf隔离，但本质上还是使用的同一块内存

2. duplicate

零拷贝的体现之一，就好比截取了原始 ByteBuf 所有内容，并且没有 max capacity 的限制，也是与原始 ByteBuf 使用同一块底层内存，只是读写指针是独立的

@Test
public void testDuplicate(){
    ByteBuf origin = ByteBufAllocator.DEFAULT.buffer(10);
    origin.writeBytes(new byte[]{0, 1, 2, 3, 4, 5, 6, 7, 8, 9});

    //拷贝一个新对象，在新对象上修改不会影响前对象
    ByteBuf duplicate = origin.duplicate();
    log(duplicate);

    //在origin的基础上新写入内容不会改变origin的内容
    duplicate.writeByte(10);
    //在origin原有的内容上修改则会改变origin的内容
    duplicate.setByte(0, 1);
    log(duplicate);
    log(origin);
}

打印结果

         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 00 01 02 03 04 05 06 07 08 09                   |..........      |
+--------+-------------------------------------------------+----------------+
read index:0 write index:11 capacity:64
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 01 02 03 04 05 06 07 08 09 0a                |...........     |
+--------+-------------------------------------------------+----------------+
read index:0 write index:10 capacity:64
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 01 02 03 04 05 06 07 08 09                   |..........      |
+--------+-------------------------------------------------+----------------+

3. copy

会将底层内存数据进行深拷贝，因此无论读写，都与原始 ByteBuf 无关

4. CompositeByteBuf

零拷贝的体现之一，可以将多个 ByteBuf 合并为一个逻辑上的 ByteBuf，避免拷贝

有两个 ByteBuf 如下

ByteBuf buf1 = ByteBufAllocator.DEFAULT.buffer(5);
buf1.writeBytes(new byte[]{1, 2, 3, 4, 5});
ByteBuf buf2 = ByteBufAllocator.DEFAULT.buffer(5);
buf2.writeBytes(new byte[]{6, 7, 8, 9, 10});

log(buf1);
log(buf2);

现在需要一个新的 ByteBuf，内容来自于刚才的 buf1 和 buf2，如何实现？

方法一

ByteBuf buf3 = ByteBufAllocator.DEFAULT.buffer(buf1.readableBytes()+buf2.readableBytes());
buf3.writeBytes(buf1);
buf3.writeBytes(buf2);
System.out.println(ByteBufUtil.prettyHexDump(buf3));

打印结果

read index:0 write index:10 capacity:10
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03 04 05 06 07 08 09 0a                   |..........      |
+--------+-------------------------------------------------+----------------+

这种方法好不好？回答是不太好，因为进行了数据的内存复制操作

方法二

CompositeByteBuf buf4 = ByteBufAllocator.DEFAULT.compositeBuffer();
// true 表示增加新的 ByteBuf 自动递增 write index, 否则 write index 会始终为 0
buf4.addComponents(true, buf1, buf2);
log(buf4);

打印结果

read index:0 write index:10 capacity:10
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03 04 05 06 07 08 09 0a                   |..........      |
+--------+-------------------------------------------------+----------------+

注意：被writeBytes()调用过的ByteBuf无法被添加到CompositeByteBuf中

总结

CompositeByteBuf 是一个组合的 ByteBuf，它内部维护了一个 Component 数组，每个 Component 管理一个 ByteBuf，记录了这个 ByteBuf 相对于整体偏移量等信息，代表着整体中某一段的数据

优点：对外是一个虚拟视图，组合这些 ByteBuf 不会产生内存复制
缺点：复杂了很多，多次操作会带来性能的损耗

5. Unpooled

Unpooled 是一个工具类，类如其名，提供了非池化的 ByteBuf 创建、组合、复制等操作
这里仅介绍其跟【零拷贝】相关的 wrappedBuffer 方法，可以用来包装 ByteBuf

ByteBuf buf1 = ByteBufAllocator.DEFAULT.buffer(5);
buf1.writeBytes(new byte[]{1, 2, 3, 4, 5});
ByteBuf buf2 = ByteBufAllocator.DEFAULT.buffer(5);
buf2.writeBytes(new byte[]{6, 7, 8, 9, 10});

// 当包装 ByteBuf 个数超过一个时, 底层使用了 CompositeByteBuf
ByteBuf buf3 = Unpooled.wrappedBuffer(buf1, buf2);
System.out.println(ByteBufUtil.prettyHexDump(buf3));

输出

         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03 04 05 06 07 08 09 0a                   |..........      |
+--------+-------------------------------------------------+----------------+

也可以用来包装普通字节数组，底层也不会有拷贝操作

ByteBuf buf4 = Unpooled.wrappedBuffer(new byte[]{1, 2, 3}, new byte[]{4, 5, 6});
System.out.println(buf4.getClass());
System.out.println(ByteBufUtil.prettyHexDump(buf4));

输出

class io.netty.buffer.CompositeByteBuf
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03 04 05 06                               |......          |
+--------+-------------------------------------------------+----------------+

10. ByteBuf优势

池化 - 可以重用池中 ByteBuf 实例，更节约内存，减少内存溢出的可能
读写指针分离，不需要像 ByteBuffer 一样切换读写模式
可以自动扩容
支持链式调用，使用更流畅
很多地方体现零拷贝，例如 slice、duplicate、CompositeByteBuf

11. 双向通信

练习案例代码

服务端

@Slf4j
public class communicationServer {
    public static void main(String[] args) {
        new ServerBootstrap()
                .group(new NioEventLoopGroup())
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) {
                        ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) {
                                ByteBuf buffer = (ByteBuf) msg;
                                log.debug("server：" + buffer.toString(Charset.defaultCharset()));

                                // 建议使用 ctx.alloc() 创建 ByteBuf
                                ByteBuf response = ctx.alloc().buffer();
                                response.writeBytes(buffer);
                                ctx.writeAndFlush(response);

                                // 思考：需要释放 buffer 吗
                                while (!buffer.release()){}
                                // 思考：需要释放 response 吗
                                while (!response.release()){}
                            }
                        });
                    }
                }).bind(8080);
    }
}

客户端

@Slf4j
public class communicationClient {
    public static void main(String[] args) throws InterruptedException {
        NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        Channel channel = new Bootstrap()
                .group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new StringEncoder());
                        ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                ByteBuf buffer = (ByteBuf) msg;
                                log.debug("client：" + buffer.toString(Charset.defaultCharset()));
                            }
                        });
                    }
                })
                .connect("localhost", 8080).sync().channel();

        channel.closeFuture().addListener(future -> {
            eventLoopGroup.shutdownGracefully();
        });

        new Thread(() -> {
            System.out.print("请输入");
            Scanner scanner = new Scanner(System.in);
            while (true) {
                String line = scanner.nextLine();
                if ("q".equals(line)) {
                    channel.close();
                    break;
                }
                channel.writeAndFlush(line);
            }
        }).start();
    }
}

💡 读和写的误解

最初在认识上有这样的误区，认为只有在 netty，nio 这样的多路复用 IO 模型时，读写才不会相互阻塞，才可以实现高效的双向通信，但实际上，Java Socket 是全双工的：在任意时刻，线路上存在A 到 B 和 B 到 A 的双向信号传输。即使是阻塞 IO，读和写是可以同时进行的，只要分别采用读线程和写线程即可，读不会阻塞写、写也不会阻塞读

例如：

public class TestServer {
    public static void main(String[] args) throws IOException {
        ServerSocket ss = new ServerSocket(8888);
        Socket s = ss.accept();

        new Thread(() -> {
            try {
                BufferedReader reader = new BufferedReader(new InputStreamReader(s.getInputStream()));
                while (true) {
                    System.out.println(reader.readLine());
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }).start();

        new Thread(() -> {
            try {
                BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(s.getOutputStream()));
                // 例如在这个位置加入 thread 级别断点，可以发现即使不写入数据，也不妨碍前面线程读取客户端数据
                for (int i = 0; i < 100; i++) {
                    writer.write(String.valueOf(i));
                    writer.newLine();
                    writer.flush();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }).start();
    }
}

客户端

public class TestClient {
    public static void main(String[] args) throws IOException {
        Socket s = new Socket("localhost", 8888);

        new Thread(() -> {
            try {
                BufferedReader reader = new BufferedReader(new InputStreamReader(s.getInputStream()));
                while (true) {
                    System.out.println(reader.readLine());
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }).start();

        new Thread(() -> {
            try {
                BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(s.getOutputStream()));
                for (int i = 0; i < 100; i++) {
                    writer.write(String.valueOf(i));
                    writer.newLine();
                    writer.flush();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }).start();
    }
}

二、Netty入门

一、概述

1. 什么是Netty

1. 简介

2. Netty的地位

3. Netty的优势

2. Netty和Tomcat有什么区别

3. 为什么Netty受大众欢迎

4. Netty为什么并发高

5. Netty为什么传输快

二、入门案例

1. 目标

2. 环境准备

3. 服务端

4. 客户端

5. 流程数量

6. 💡提示

三、组件

1. EventLoop

1. 概述

EventLoop：事件循环对象

EventLoopGroup：事件循环组

2. 案例演示EventLoop执行

3 💡 优雅关闭

4. 演示 NioEventLoop 处理 io 事件

案例

改进点一

改进点二

5. 💡 handler 执行中如何换人？

2. Channel

1. 常用方法

2. ChannelFuture

3. CloseFuture

4. 💡 异步提升的是什么

3. Future & Promise

1. 概述

2. JDK Future

3. Netty Future

4. Netty Promise

案例一：同步处理任务

案例二：异步处理任务

案例三：同步处理任务失败

Sync

Await

案例四：异步处理任务失败

案例五：Await死锁

5. Handler & Pipeline

ctx.channel().write(msg) VS ctx.write(msg)

6. EmbeddedChannel基本使用

四、ByteBuf

1. 创建

2. 直接内存 vs 堆内存

3. 池化 vs 非池化

4. ByteBuf的组成

5. 写入

6. 扩容

7. 读取

8. retain & release

9. ByteBuf零拷贝

1. Slice

2. duplicate

3. copy

4. CompositeByteBuf

方法一

方法二

总结

5. Unpooled

10. ByteBuf优势

11. 双向通信

练习案例代码

服务端

客户端

💡 读和写的误解