Netty(2) 入门篇 | 核心概念理解一臂之力

前言

本文开始正式对 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.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 的 readwrite 事件（OP_READOP_WRITE 事件）
• ② step2-processSelectedKeys：在对应的 SocketChannel 上处理 readwrite 事件
• ③ 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

加入依赖

    
        
            io.netty
            netty-all
            4.1.39.Final
        
    

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() {
                        @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() {
                    @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 extends java.util.concurrent.Future 

public interface Promise extends Future

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 future = pool.submit(new Callable() {
            @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 future = eventLoop.submit(new Callable() {

            @Override

            public Integer call() throws Exception {

                log.debug("执行计算");

                Thread.sleep(1000);

                return 50;

            }

        });

        log.debug("等待结果");

//        log.debug("结果是 {} ",future.get());

        future.addListener(new GenericFutureListener