前言

本文开始正式对 Netty 的相关概念进行学习，从 Netty 的概述、线程模型、到入门代码，并介绍 Netty 的核心组件，完成对 Netty 的基础学习阶段。

一、概述

1.1 介绍

Netty 是一个异步的，基于事件驱动的网络应用框架，用于快速开发可维护、高性能的网络服务器和客户端。这里的异步，并不是使用 AIO，还是使用多路复用模型。

netty.io/

Netty是由 JBOSS 提供的一个java开源框架，现为 Github上的独立项目。Netty提供非阻塞的、事件驱动的网络应用程序框架和工具，用以快速开发高性能、高可靠性的网络服务器和客户端程序。作者如下，韩国人。

Netty vs 其他网络应用框架

• Mina 由 apache 维护，将来 3.X 版本可能会有较大重构，破坏 API 向下兼容性，Netty 的开发迭代更迅速，API 更简介，文档更优秀。
• 久经考验、出来19年
• netty 版本

• • 2.x 2004
  • 3.x 2008
  • 4.x 2013
  • 5.x 已废弃（没有明显的性能提升，维护成本高）

1. Netty 版本分为 Netty 3.x 和 Netty 4.x、Netty 5.x
2. 因为 Netty 5 出现重大 bug，已经被官网废弃了，目前推荐使用的是 Netty 4.x的稳定版本
3. 目前在官网可下载的版本 Netty 3.x、Netty 4.0.x 和 Netty 4.1.x

本质：网络应用程序框架；

实现：异步、事件驱动

特性：高性能、可维护、快速开发

用途：开发服务器和客户端

1.2 地位

Netty 在 Java 网络中的地位，相当于 Spring 在 JavaEE 开发中的地位。

Netty 是目前最流行的 NIO 框架，Netty 在互联网领域、大数据分布式计算领域、游戏行业、通信行业获得了广泛的应用，以下框架都使用了 Netty，因为他们有网络通讯需求。

• Cassandra nosql 数据库
• Spark
• Hadoop
• RocketMQ
• Es
• Dubbo
• Spring5.x
• zookeeper

1.3 Netty 的优势

网络开发应用为什么不选 JDK 原生API，而选用 Netty 的理由。

① Netty 的 API 比原生的 API 更强大

• JDK 中的 NIO 的一些 API 功能薄弱且复杂，Netty 隔离了 JDK 中的 NIO 的实现变化和细节，例如：ByteBuffer -> 改成 ByteBuf，主要负责从底层的 IO 中读取数据到 ByteBuf，然后传递给应用程序，应用程序处理完之后封装为 ByteBuf，写回给 IO

② Netty 自身线程安全

• 使用 JDK 原生 API 需要对多线程非常熟悉，因为 NIO 涉及到 Reactor 设计模式，而 Netty 帮我们封装好了这层，我们只需要考虑 Handler 的编写问题

③ 完整的高可用机制

• JDK 原生方式要实现高可用，需要自己来处理很多异常，比如网路断路重连、粘包处理、失败缓存的处理等，而 Netty 则做的更多，它解决了传输的一些问题，比如粘包半包问题，它支持常用的应用层协议，完整的断路重连，idle 等异常处理。

④ JDK bug

• JDK 的 NIO 中有个 Bug，例如 Epoll，它会导致 CPU 100% 空轮询。

1.4 架构

1.4.1 Netty 功能

核心：

• 可扩展的事件模型
• 统一的通信 API，简化了通信编码
• 零拷贝机制与丰富的字节缓冲区

传输服务：

• 支持 socket 以及 datagram（数据包）
• http 传输服务
• In-VM Pipe（管道协议，是jvm的一种进程）

协议支持：

• http 以及 websocket
• SSL 安全套接字协议支持
• Google Protobuf （序列化框架）
• 支持zlib、gzip压缩
• 支持大文件的传输
• RTSP（实时流传输协议，是TCP/IP协议体系中的一个应用层协议）
• 支持二进制协议并且提供了完整的单元测试

1.4.2 线程模型

前面分析过，Netty 的线程模型是基于 Reactor 模型实现的，对 Reactor 三种模式都有非常好的支持，并做了一定的改进，一般情况下，在服务端会采用主从架构模型。

简易版本：

对比上图：

• 1、BoosGroup 线程维护 Selector，只关注 Accepet，相当于主 Reactor，只关注连接事件。
• 2、当接收到 Accept 事件，获取到对应的 SocketChannel，封装成 NIOSocketChannel 并注册到 Worker 线程（从Reactor）（事件循环），并进行维护
• 3、当 Worker 线程监听到 Selector 中通道发生自己感兴趣的事件后，就进行处理（handler）

详细版本图：

再来详细的捋一遍 Netty 的工作流程。

1、Netty 抽象出两组线程池：BossGroup 和 WorkerGroup ，每个线程池中都有 EventLoop 线程（可以是BIO、NIO、AIO）。

• BossGroup 中的线程专门负责和客户端建立连接，相当于老板线程在与客户建立关系拉项目做；
• WorkerGroup 中的线程专门负责处理连接上的读写，相当于打工人线程在拼命的做拉过来的项目；

2、EventLoop 表示一个不断循环的执行事件处理的线程，每个 EventLoop 都包含一个 Selector ，用于监听注册在它上面的 Socket 网络连接（Channel），多个 NioEventLoop 组成 EventLoopGroup，相当于事件循环组，这个组中有很多项目在忙。

3、每个 Boos EventLoop 中循环执行以下三个步骤：

• ① step1-select：轮询注册在其上的 ServerSocketChannel 的 accept 事件（OP_ACCEPT 事件）
• ② step2- processSelectedKeys ：处理 accept 事件，与客户端建立连接，生成一个 SocketChannel，并将其注册到某个 worker EventLoop 的 Selector 上
• ③ step3- runAllTasks ： 再去以此循环处理队列中的其他任务

4、每个 Worker EventLoop 中循环执行以下三个步骤：

• ① step1-select：轮询注册在其上的 ServerSocketChannel 的 read\write 事件（OP_READ\OP_WRITE 事件）
• ② step2-processSelectedKeys：在对应的 SocketChannel 上处理 read\write 事件
• ③ step3-runAllTasks：再去以此循环处理任务队列中的其他任务

5、在以上两个 processSelectedKeys 步骤中，会使用 Pipeline（管道），Pipeline 中引用了 Channel，即可通过 Pipeline 可以获取到对应的 Channel，Pipeline 中维护了很多的处理器（拦截处理器、过滤处理器、自定义处理器）。

总结：

1、Netty 的线程模型基于主从 Reactor 模型，通常由一个线程负责处理 OP_ACCEPT 事件，拥有 CPU 核数的两倍 IO 线程处理读写事件；

2、一个通道的 IO 操作会绑定在一个 IO 线程中，而一个 IO 线程可以注册多个通道；

3、在一个网络通信中通常会包含网络数据读写，编码，解码，业务处理。默认情况下网络读写、编码、解码等操作会在 IO 线程中运行，但也可以指定其他线程池。

4、通常业务处理会单独开启业务线程池（看业务类型），但也可以进一步细化，例如心跳包可以直接在 IO 线程中处理，而需要再转发给业务线程池，避免线程切换；

5、在一个 IO 线程中所有的通道的事件是串行处理的。

6、通常业务操作会专门开辟一个线程池，那业务处理完成之后，如果将响应结果通过 IO 线程写入到网卡中呢？业务线程调用 Channel 对象的 write 方法并不会立即写入网络，只是将数据放入一个待写入缓存区，然后 IO 线程每次执行事件选择后，会从待写入缓存区中获取写入任务，将数据真正写入到网络中。

二、经典的 Helloworld

加入依赖

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

2.1 编写 servre

package com.xiaolei.netty.helloreactor;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;

/**
 * @author xiaolei
 * @version 1.0
 * @date 2023-08-19 11:46
 */
public class RHelloServer {
    public static void main(String[] args) {
        RHelloServer server = new RHelloServer();
        server.start(8888);
    }

    private void start(int port){
        NioEventLoopGroup boss = new NioEventLoopGroup(1);
        NioEventLoopGroup work = new NioEventLoopGroup();
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(boss,work) // 配置 group
                    .channel(NioServerSocketChannel.class) // 服务端 channel
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel channel) throws Exception {
                            // 6. 添加具体的 handler
                            channel.pipeline().addLast(new ServerHandler1());
                        }
                    });
            // 绑定端口启动服务端
            ChannelFuture future = bootstrap.bind(port).sync();

            // 等待端口启动服务端,监听服务端的关闭事件
            future.channel().closeFuture().sync();
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            boss.shutdownGracefully();
            work.shutdownGracefully();
        }
    }
}

2.2 编写 client

package com.xiaolei.netty.helloreactor;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;

import java.nio.charset.StandardCharsets;

/**
 * @author xiaolei
 * @version 1.0
 * @date 2023-08-19 12:12
 */
public class RNettyClient {
    public static void main(String[] args) throws InterruptedException {
        RNettyClient client = new RNettyClient();
        client.connect("127.0.0.1",8888);
    }

    public void connect(String host,int port) throws InterruptedException {
        // 1. 启动类
        ChannelFuture future = new Bootstrap()
                // 2. 添加 EventLoop
                .group(new NioEventLoopGroup())
                // 3. 选择客户端 channel 实现
                .channel(NioSocketChannel.class)
                // 4. 添加处理器, 处理器在连接建立后调用，执行初始化
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel channel) throws Exception {
                        channel.pipeline().addLast(new ClientHandler1());
                    }
                })
                .connect(host, port)
                .sync();// 阻塞方法，直到连接建立

        // 连接建立，就可以发送数据了
        Channel channel = future.channel();
        ByteBuf buffer = channel.alloc().buffer();
        buffer.writeBytes("hello world,i am a netty client".getBytes(StandardCharsets.UTF_8));
        channel.writeAndFlush(buffer);

        future.channel().closeFuture().sync();
    }
}

2.3 编写 Handler

服务端 Handler

package com.xiaolei.netty.helloreactor;

import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.socket.SocketChannel;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;

/**
 * @author xiaolei
 * @version 1.0
 * @date 2023-08-20 14:48
 */
@Slf4j
@ChannelHandler.Sharable
public class ServerHandler1 extends ChannelInboundHandlerAdapter{

    public ServerHandler1() {
        super();
    }

    /**
     * channel 注册事件
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        log.info("registered 事件");
        super.channelRegistered(ctx);
    }

    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        log.info("unregistered 事件");
        super.channelUnregistered(ctx);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info("channel active 事件");
        super.channelActive(ctx);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        log.info("channel 连接关闭事件");
        super.channelInactive(ctx);
    }

    /**
     * channel 中读取到数据了
     * @param ctx
     * @param msg
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        log.info("channel read 事件");
        ByteBuf buf = (ByteBuf)msg;
        byte[] bytes = new byte[buf.readableBytes()];
        buf.readBytes(bytes);

        String message = new String(bytes, Charset.defaultCharset());
        log.info("server 读取的数据是 ：{}",message);
        super.channelRead(ctx, msg);
    }

    /**
     * 通道内数据完全读完了
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        System.out.println("进入 inboud1，给客户端发送数据");

        // 给客户端做一个响应
        ByteBuf buffer = ctx.alloc().buffer();
        buffer.writeBytes("hello cleint,我读完了".getBytes(StandardCharsets.UTF_8));
        ctx.writeAndFlush(buffer);
        // 写的数据，会经过 服务端的 out 类型的 handler

        super.channelReadComplete(ctx);
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        super.userEventTriggered(ctx, evt);
    }

    @Override
    public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
        super.channelWritabilityChanged(ctx);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        super.exceptionCaught(ctx, cause);
    }
}

客户端Handler

package com.xiaolei.netty.helloreactor;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;

/**
 * @author xiaolei
 * @version 1.0
 * @date 2023-08-20 14:48
 */
@Slf4j
@ChannelHandler.Sharable
public class ClientHandler1 extends ChannelInboundHandlerAdapter{

    public ClientHandler1() {
        super();
    }

    /**
     * channel 注册事件
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        log.info("registered 事件");
        super.channelRegistered(ctx);
    }

    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        log.info("unregistered 事件");
        super.channelUnregistered(ctx);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info("channel active 事件");
        super.channelActive(ctx);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        log.info("channel 连接关闭事件");
        super.channelInactive(ctx);
    }

    /**
     * channel 中读取到数据了
     * @param ctx
     * @param msg
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        log.info("channel read 事件");
        ByteBuf buf = (ByteBuf)msg;
        byte[] bytes = new byte[buf.readableBytes()];
        buf.readBytes(bytes);

        String message = new String(bytes, Charset.defaultCharset());
        log.info("client 读取的数据是 ：{}",message);
        super.channelRead(ctx, msg);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        super.channelReadComplete(ctx);
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        super.userEventTriggered(ctx, evt);
    }

    @Override
    public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
        super.channelWritabilityChanged(ctx);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        super.exceptionCaught(ctx, cause);
    }
}

先启动服务端，然后启动客户端。

在本例的 helloworld 中 引用了两个 handler ，一个服务端的handler，一个客户端的 handler ，分别继承 ChannelInboundHandlerAdapter 方法，实现它的 channelRead 和 channelReadComplete 事件，代表读取事件和 读取完毕后的事件。

在这个例子中，我们要树立正确的观念。

• 把 channel 理解为数据的通道
• 把 msg 理解为 流动的数据，最开始输入是 ByteBuf，但是经过 pipeline 的加工，会变成其他类型对象，最后输出又变成 ByteBuf，也就是 ByteBuf 是数据传输的容器。
• 把 handler 理解为数据的处理工具

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

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

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

Netty 的大部分操作都是异步的，比如地址绑定，客户端连接等。

对应我们的线程模型，我们可以在 netty 中这样写

NioEventLoopGroup boss = new NioEventLoopGroup(1);
NioEventLoopGroup work = new NioEventLoopGroup();

ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(boss,work) // 配置 group

三、Netty 组件

3.1 EventLoop

从上面的线程模型中，可以看到 bootstrap，配置了两个主从 EventLoopGroup。

其中 Bootstrap 是引导的意思，作用是配置整个 Netty 程序，将各个组件都串起来，最后绑定端口，启动 Netty 服务。

Netty 中提供了 2 种类型的引导类，一种用于客户端（Bootstrap），而另一种（ServerBootStrap）用于服务器。

• ServerBootStrap 将绑定到一个端口，因为服务器必须监听连接，而 BootStrap 则作为客户端来连接服务器使用
• 一个客户端的 Bootstrap 只要一个 EventLoopGroup，而 Server 则要两个。

Netty 是基于事件驱动的，比如上面 handler 的重写方法种的：连接注册事件、连接激活、数据读取、异常事件等，有了事件，就需要一个组件去监听事件的产生和事件的处理，这个组件就是 EventLoop（事件循环\EventExecutor）

在 Netty 中，每个 Channel 都会被分配到一个 EventLoop，一个 EventLoop 可以服务于多个 Channel，每个 EventLoop 会占用一个 Theard，同时这个 Thread 会处理 EventLoop 上面发生的所有 IO 操作和事件。

EventLoopGroup 是用来生成 EventLoop 的，包含了一组 EventLoop（可以初步理解成线程池）。

EventLoop 本质是一个单线程执行器，里面维护了一个 Selector，里面有 run 方法处理 Channel 上源源不断的 IO 事件。

它的继承关系比较复杂

• 一条线是继承自 j.u.c.ScheduledExecutorService 因此包含了线程池中所有方法
• 令一条线是继承自 netty 自己的 OrderedEventExecutor

• • 提供了 boolean inEventLoop（Thread thread）方法判断一个线程是否属于此 EventLoop
  • 提供了 parent 方法来看看自己属于哪个 EventLoopGroup

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

• 继承自 netty 自己的 EventExecutorGroup

• • 实现了 Iterable 接口提供遍历 EventLoop 的能力
  • 另有 next 方法获取集合中下一个 EventLoop

3.2 Channel

Netty 的 Channel 种类不多，

• NioServerSocketChannel：通用的NIO通道模型，也是Netty的默认通道。
• EpollServerSocketChannel：对应Linux系统下的epoll多路复用函数。
• KQueueServerSocketChannel：对应Mac系统下的kqueue多路复用函数。
• OioServerSocketChannel：对应原本的BIO模型，用的较少，一般用原生的。

channel 的主要作用

• close 可以用来关闭 channel
• closeFuture（）用来处理 channel 的关闭

• • sync 方法作用是同步等待 channel 关闭
  • 而 addListener 方法是异步等待 channel 关闭

• pipeline 方法添加处理器（添加工序处理）
• write 方法将数据写入channel，（但不会立刻发出，有很多条件才能发出）
• writeAndFlush 方法将数据写入并刷出（数据写入channel，并立刻发出）

建立 channel 的是 nio 线程，connect 是异步非阻塞，main 发起了调用。

所以用 sync 同步等待 channel 。

小结：主要作用呢:

1、通过 Channel 可获得当前网络连接的通道状态

2、通过 Channel 可获得网络连接的配置参数（缓冲区大小）

3、Channel 提供异步的网络 IO 操作

Netty 中的很多方法都采用了异步，不能同步处理，例如 connect、sync，这个并不是 Netty 效率高的原因，更重要的原因是 Netty 采用了流水线的设计，分工更细。

3.3 Future & Promise

在异步处理时，经常用到这两个接口。

首先要说明 netty 中的 Future 与 jdk 的Future 同名，但是两个接口，netty 的 Future 继承自 jdk 的 Future，而 Promise 又对 netty Future 进行了扩展。

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

public interface Future<V> extends java.util.concurrent.Future<V> 

public interface Promise<V> extends Future<V>

netty 的 Future

• getNow：获取任务结果，非阻塞，还未产生结果时返回 null
• await ： 等待任务结束，如果任务失败，不会抛出异常，而是通过 isSuccess 判断
• sync ： 等待任务结束，如果任务失败，抛出异常
• isSucess ： 判断任务是否成功
• cause：获取失败信息，非阻塞，如果没有失败，返回 null
• addLinstener：添加回调，异步接收结果

Promise

• setSuccess
• setFailure

3.3.1 JDK-Future

@Slf4j
public class TestJdkFuture {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService pool = Executors.newFixedThreadPool(2);
        Future<Integer> future = pool.submit(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                log.debug("执行计算");
                Thread.sleep(1000);
                return 50;
            }
        });
        log.debug("等待结果");
        log.debug("结果是 {}",future.get());
    }
}

3.3.2 Netty-Future

异步方式获取结果

@Slf4j
public class TestNettyFuture {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        NioEventLoopGroup group = new NioEventLoopGroup(2);

        EventLoop eventLoop = group.next();

        Future<Integer> future = eventLoop.submit(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                log.debug("执行计算");
                Thread.sleep(1000);
                return 50;
            }
        });
        log.debug("等待结果");
//        log.debug("结果是 {} ",future.get());
        future.addListener(new GenericFutureListener<Future<? super Integer>>() {
            @Override
            public void operationComplete(Future<? super Integer> future) throws Exception {
                log.debug("结果是 {} ",future.getNow());
            }
        });
    }
}

Future \ Promise 异步模型

• future 和 promise，目的是将值（future）与其计算方式（promise）分离，从而允许更灵活的进行计算，特别是通过并行化。Future 表示目标计算的返回值，Promise 表示计算的方式，这个模型将返回结果和计算逻辑分离，目的是为了让计算逻辑不影响返回结果，从而抽象出一套异步编程模型。而计算逻辑与结果关联的纽带就是 callback
• Netty 中有非常多的异步调用，譬如：client\server 的启动，连接、数据的读写等操作都是支持异步的。

3.3.3 Promise

Netty 的 Future，只是增加了监听器，整个异步的状态，是不能进行设置和修改的，于是 Netty 的 Promise 接口扩展了 Netty 的 Future 接口，可以设置异步执行的结果。在 IO 操作过程，如果顺利完成，或者发生异常，都可以设置 Promise 的结果，并且通知 Promise 的 Listener 们。

在 Java 的 Future 中，业务逻辑为一个 Callable 或 Runnable 实现类，该类的 call() 或 run()执行完毕意味着业务逻辑的完结，在 Promise 机制中，可以在业务逻辑中人工设置业务逻辑的成功与失败，这样更加方便的监控自己的业务逻辑。

package com.xiaolei.netty.common;

import io.netty.channel.EventLoop;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.util.concurrent.DefaultPromise;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ExecutionException;

/**
 * @author xiaolei
 * @version 1.0
 * @date 2023-06-18 10:23
 */
@Slf4j
public class TestNettyPromise {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 1. 准备 EventLoop 对象
        EventLoop eventLoop = new NioEventLoopGroup().next();

        DefaultPromise<Integer> promise = new DefaultPromise<>(eventLoop);

        new Thread(()->{
            // 任意一个线程执行计算，计算完毕后向 promise 填充结果
            System.out.println("开始计算....");
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            promise.setSuccess(100);
        }).start();
        // 接收结果的线程
        log.debug("watting ");
        log.debug("result {}",promise.get());

    }
}

ChannelPromise 接口，则继承扩展了 Promise 和 ChannelFuture。

所以，ChannelPromise 既绑定了 Channel，又具备了设置监听回调的功能，还可以设置 IO 操作的结果，是 Netty 实际编程使用的最多的接口。

3.4 Handler & Pipeline

ChannelPipeline 提供了 ChannelHandler 链的容器，以服务端程序为例，

• 客户端发送过来的数据要接收，读取处理，我们称数据是入站的，需要经过一系列 Handler 处理后；
• 如果服务器想向客户端写回数据，也需要经过一系列 Handler 处理，我们称数据是出战的。

pipeline 好比流水线，Handler 就是处理器

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

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

入站指返回响应，而 Netty 的出站则是指返回相应。

注意：netty 会默认加两个 handler ，一个是 head 一个是 tail ，我们使用 addLast 并非是在最后一个，而是加在 tail 之前。

• 当接收到客户端的数据时，Netty 会从 Head 节点开始依次往后执行所有入栈处理器。
• 而当服务端返回数据时，Netty 会从 Tail 节点开始依次向前执行所有入栈处理器。

pipe -> head -> h1 -> h2 -> h3 -> h4 -> h5 ->h6 -> tail 。

3.4.1 Inbound-handler

流水线的设计，就是可以将数据进行处理后，将结果传给下一个 handler，将数据经过多次工序。

传递方式： super.channelRead(ctx, msg);

下一个 handler 拿取方式，直接从方法参数 msg 中拿到

如果在，h2 中不调用 channelRead ，那么链路就会断掉。

3.4.2 outbound-handler

• InboundHandler是按照Pipleline的加载顺序(addLast)， 顺序 执行
• OutboundHandler是按照Pipeline的加载顺序(addLast)， 逆序 执行

出站处理器是在服务端返回数据时被触发的，如果在入站处理器中加行代码，向客户端发送数据，就会触发。

问题：回显数据事件流转规则？

• 如果是通过 Channel 对象进行数据回写，事件会从 pipeline 尾部流向头部（见上图）
• 如果是通过 ChannelHandlerContext 对象进行数据回写，事件会从当前 handler 流向头部
• 问题：OutboundHandler 和 InboundHandler 的先后顺序是否有要求？才能保证所有 outboundHandler 能被执行

• • 如果想让所有的 OutboundHandler 都能被执行到，可以选择把 OutboundHandler 放在最后一个有效的 InboundHandler 之前
  • 有一种做法是通过 addFirst 加载所有 OutboundHandler，再通过 addLast 加载所有 InboundHandler；
  • 另外也推荐：使用 addLast 先加载所有 OutboundHandler，然后加载所有 InboundHandler

注意：在 outboundhandler 中最好不要通过 Channel 写数据，会导致事件再次从尾部流动到头部，造成类似递归问题。

可以在事件向前传播出去之后通过 ChannelHandlerContext 写数据。

3.4.3 自定义出站入站处理器

实际开发过程中，通常 pipeline.addLast 并不会直接 new 接口，而是自己定义处理器类，然后继承对应的父类，如上面我们的 Handler 的创建

对于入站处理器而言，主要重写其 channelRead（）方法即可，出战处理器，就重写 write 方法即可，这样所有消息出站时，都会调用该方法

最后，不仅仅处理器可以单独抽出来实现，而且对于通道的初始化器，也可也单独抽出来实现，如下：

package com.xiaolei.netty.bosschat.websocket;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpServerCodec;

/**
 * @author xiaolei
 * @version 1.0 channel 注册后，会执行里面相应的初始化方法
 * @date 2023-07-22 17:59
 */
public class HttpServerInitHandler extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel channel) throws Exception {
        // 通过 SocketChannel 获取 pipeline
        ChannelPipeline pipeline = channel.pipeline();
        /**
         * HttpServerCodec ： 是由 Netty 自己提供的助手类，编解码工作
         */
        pipeline.addLast("HttpServerCodec",new HttpServerCodec());
        pipeline.addLast("httpHandler",new HttpHandler());
    }
}

// 服务端调用

           ChannelFuture channelFuture = serverBootstrap.group(bossGroup, workGroup)
                    .channel(NioServerSocketChannel.class) // BIO、NIO 这里选择 NIO
                    .childHandler(new HttpServerInitHandler())
                    .bind(875).sync();

3.4.4 ChannelHandler 复用

每个客户端Channel创建后初始化时， 均会向与该Channel绑定的Pipeline中 添加handler，此种模式下，每个

Channel享有的是各自独立的Handler 。

我们可以加上注解，即可保证复用问题

@ChannelHandler.Sharable

注解只是标注可被复用，至于线程安全问题，需要开发者自己保证。

3.4.5 ChannelInboundHandlerAdaper

对于编写Netty数据入站处理器，可以选择继承 ChannelInboundHandlerAdapter，也可以选择继承

SimpleChannelInboundHandler,区别是什么？

public class TestByteBuf {
    public static void main(String[] args) {
        ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer();
        System.out.println(buffer);
        StringBuilder builder = new StringBuilder();
        for (int i = 0; i < 300; i++) {
            builder.append("aa");
        }
        buffer.writeBytes(builder.toString().getBytes());
        System.out.println(buffer);
    }
}

PooledUnsafeDirectByteBuf(ridx: 0, widx: 0, cap: 256)
PooledUnsafeDirectByteBuf(ridx: 0, widx: 600, cap: 1024)

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

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

capacity()：表示 ByteBuf 底层占用了多少字节的内存（包括丢弃的字节、可读字节、可写字节），不同的底层实
现机制有不同的计算方式。

maxCapacity()： ByteBuf 底层最大能够占用多少字节的内存，当向 ByteBuf 中写数据的时候，如果发现容量不
足，则进行扩容，直到扩容到 maxCapacity，超过这个数，就抛异常。

readableBytes() 与 isReadable()：readableBytes() 表示 ByteBuf 当前可读的字节数，它的值等于
writerIndex-readerIndex，如果两者相等，则不可读，isReadable() 方法返回 false

writableBytes()、 isWritable() 、maxWritableBytes()：writableBytes() 表示 ByteBuf 当前可写的字节数，它的
值等于 capacity()-writerIndex，如果两者相等，则表示不可写，isWritable() 返回 false，但是这个时候，并不代
表不能往 ByteBuf 中写数据了， 如果发现往ByteBuf 中写数据写不进去的话，Netty 会自动扩容 ByteBuf，直到扩
容到底层的内存大小为 maxCapacity，而 maxWritableBytes() 就表示可写的最大字节数，它的值等于
maxCapacity-writerIndex。

readerIndex() 与 readerIndex(int readerIndex)：前者表示返回当前的读指针 readerIndex, 后者表示设置读指针

writeIndex() 与 writeIndex(int writerIndex)：前者表示返回当前的写指针 writerIndex, 后者表示设置写指针

markReaderIndex() 与markWriterIndex()：表示把当前的读指针/写指针保存起来，操作形式为：
markedReaderIndex = readerIndex / markedWriterIndex = writerIndex;

writeBytes(byte[] src)： 表示把字节数组 src 里面的数据全部写到 ByteBuf，src字节数组大小的长度通常小于等于
writableBytes()

readBytes(byte[] dst)：把 ByteBuf 里面的数据全部读取到 dst，dst 字节数组的大小通常等于 readableBytes()

writeByte(int value)、readByte()：writeByte() 表示往 ByteBuf 中写一个字节，而 readByte() 表示从 ByteBuf 中读

取一个字节，类似的 API 还有 writeBoolean()、writeChar()、writeShort()、writeInt()、writeLong()、
writeFloat()、writeDouble() 与 readBoolean()、readChar()、readShort()、readInt()、readLong()、
readFloat()、readDouble() 等等

discardReadBytes()： 丢弃已读取的字节空间，可写空间变多

clear()：重置readerIndex 、 writerIndex 为0，需要注意的是，重置并没有删除真正的内容

release()：真正去释放bytebuf中的数据，

ReferenceCountUtil.release(buf)：工具方法，内部还是调用release()

ByteBuf buf = ...
try {

} finally {
  buf.release();
}