Netty框架详解

一、Netty简介

Netty是一款基于Java NIO的网络编程、高性能、异步事件驱动的网络应用框架。它的设计目标是提供简单易用、高性能、可扩展的网络编程框架。

二、Netty主要特点

高并发：Netty使用异步的、非阻塞的I/O模型，通过事件驱动的方式处理网络操作，能够高效地处理并发连接和大量的并发请求。
高性能：Netty采用了一系列优化策略，如零拷贝技术、内存池和可定制的线程模型等，以提供出色的性能和吞吐量。
多协议支持：Netty提供了丰富的协议支持，包括常用的网络协议（如HTTP、WebSocket、TCP和UDP）以及自定义协议。具备灵活的编解码器和处理器，简化了协议的实现和交互。
可扩展和灵活：Netty的架构和组件设计具有高度的可扩展性和灵活性。它提供了一组可重用的组件，可以根据应用需求进行定制和扩展。
安全性：Netty提供了一系列的安全性功能，包括SSL/TLS支持、加密和身份验证等。
简单易用：Netty提供了简洁的API和丰富的文档，使得开发人员可以快速上手，快速开发高质量的网络应用程序。
跨平台：Netty可以在多种操作系统平台上运行，包括Windows、Linux、Mac OS X等。

三、Netty使用场景

服务器开发：Netty可以用于开发高性能、可靠的服务器应用程序，如Web服务器、游戏服务器、消息服务器等。
客户端开发：Netty可以用于开发高性能、可靠的客户端应用程序，如聊天软件、文件传输软件等。
分布式系统：Netty可以用于开发分布式系统，如RPC框架、消息队列等。

四、Netty服务端工作架构流程

在这里插入图片描述
（1）服务端工作流程：

服务端初始化时会创建2个NioEventLoopGroup，BoosGroup用于Accept连接建立事件并分发请求，WokerGroup用于处理读写事件和业务逻辑。
服务端启动时创建 ServerBootstrap 实例，并配置 EventLoopGroup、Channel 类型和处理器，通过调用 serverBootstrap.bind() 绑定服务器端口。
BoosGroup中的NioEventLoop不断轮询注册在其Selector上的ServerSocketChannel的 Accept 事件。
接收到Accept 事件后，再交由processSelectedKeys处理 accept 事件，与客户端建立连接，生成一个 NioSocketChannel，并将其注册到WokerGroup的某个 NioEventLoop 的 Selector 上。
WokerGroup的 NioEventLoop 不断轮询注册在其Selector上的NioSocketChannel的 read/write 事件。
接收到read/write 事件后，再交由processSelectedKeys处理 read/write 事件，并调用ChannelPipeline中的相关的ChannelHandler 的channelRead()方法接收并处理客户端发送的数据。
处理完数据后，通过ChannelHandlerContext 的writeAndFlush()方法发送响应结果消息给客户端。

（2）客户端工作流程：

客户端初始化时会创建1个NioEventLoopGroup 用于处理客户端的I/O操作。
客户端启动时创建 Bootstrap 实例，并配置 EventLoopGroup、Channel 类型和处理器，通过调用 bootstrap.connect() 发起连接请求交给 NioEventLoopGroup 处理。
NioEventLoopGroup 选择一个 NioEventLoop 来处理连接请求，并与服务端建立连接。
当客户端与服务端成功建立连接时，NioEventLoop 会创建一个 Channel 对象来表示连接，该对象维护了与连接相关的状态和属性。同时客户端和服务端的 ChannelHandler 中的 channelActive() 方法会被调用。
再通过调用ChannelHandlerContext 的writeAndFlush()方法向服务端发送消息。
发送成功后，服务端接收到消息并返回处理结果时，ChannelHandler 的channelRead()方法能接收到服务端返回的响应结果消息。

五、Netty的核心组件

Bootstrap：用于启动和配置网络应用程序配置类，ServerBootstrap用于服务端，Bootstrap用于客户端。
NioEventLoopGroup：相当于一个事件循环组，这个组中含有多个事件循环，每个事件循环就是一个 NioEventLoop。在服务端会创建BossEventLoopGroup 和WorkerEventLoopGroup两个NioEventLoopGroup实例。BossEventLoopGroup 只负责处理连接事件，WorkerEventLoopGroup则负责read/write 事件。客户端通常只有一个NioEventLoopGroup来发起连接处理I/O任务。
NioEventLoop：表示一个不断循环的执行事件处理的线程，每个 NioEventLoop 都包含一个 Selector和一个TaskQueue，Selector用于监听注册在其上的 SocketChannel 上的I/O事件，如：read，write，accept，connect等；TaskQueue用于存放一些非I/O任务，如：register，bind等任务。
Channel：Netty 对网络操作抽象类，它除了包括基本的 I/O 操作，如 bind()、connect()、read()、write() 等。不同协议、不同的阻塞类型的连接都有不同的 Channel 类型与之对应。常用的类型有：NioSocketChannel（客户端 TCP Socket 连接）、NioServerSocketChannel（服务器端 TCP Socket 连接）。
ChannelHandler：是消息的具体处理器，负责处理输入输出数据的逻辑。可以接收入站事件（如数据接收）和出站事件（如数据发送），并执行相应的处理逻辑。
ChannelHandlerContext：是 ChannelHandler 的上下文环境。包含与ChannelHandler相关联的各种信息，如Channel、EventLoop、ChannelPipeline等。ChannelHandlerContext还提供了丰富的方法，以便于ChannelHandler与其他组件进行交互。
ChannelPipeline：是一个双向链表，拦截和处理事件的链式结构，主要负责管理ChannelHandler并协调它们的处理顺序。ChannelPipeline 中又维护了一个由 ChannelHandlerContext 组成的双向链表，并且每个 ChannelHandlerContext 中又关联着一个 ChannelHandler。
ChannelFuture：主要用于接收异步I/O操作返回的执行结果。ChannelFuture提供了丰富的方法，用于检查操作的状态、添加监听器以便在操作完成时接收通知，并对操作的结果进行处理。

六、TCP 粘包与拆包

拆包：是指发送方发送的一条完整的数据放入缓冲区后，接收端从缓冲区每次只读取到数据这条完整数据的一部分。
粘包：缓冲区存放着发送方发送的多条完整的数据，接收端从缓冲区每次读取到的数据是多条数据拼在一起。
之所以会产生粘包与拆包主要还是因为接收方不知道消息之间的界限，不知道一次性提取多少字节的数据所造成的。可以使用 Netty 自带的解码器来解决粘包与拆包问题。
（1）LineBasedFrameDecoder : 发送端发送数据包的时候，每个数据包之间以换行符作为分隔，LineBasedFrameDecoder 的工作原理是它依次遍历 ByteBuf 中的可读字节，判断是否有换行符，然后进行相应的截取。
（2）DelimiterBasedFrameDecoder : 可以自定义分隔符解码器，LineBasedFrameDecoder 实际上是一种特殊的 DelimiterBasedFrameDecoder 解码器。
（3）FixedLengthFrameDecoder: 固定长度解码器，它能够按照指定的长度对消息进行相应的拆包。

七、使用实例

添加 Netty 依赖

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

编写Server服务端启动类

public class NettyServer {
	 /**
     * 创建服务端实例并绑定端口
     * @throws InterruptedException
     */
    public static void bind() throws InterruptedException {
        // 创建boss线程组，用于接收连接
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        // 创建worker线程组，用于处理连接上的I/O操作，含有子线程NioEventGroup个数为CPU核数大小的2倍
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            // 创建ServerBootstrap实例，服务器启动对象
            ServerBootstrap bootstrap = new ServerBootstrap();

            // 使用链式编程配置参数
            // 将boss线程组和worker线程组暂存到ServerBootstrap
            bootstrap.group(bossGroup, workerGroup);
            // 设置服务端Channel类型为NioServerSocketChannel作为通道实现
            bootstrap.channel(NioServerSocketChannel.class);

            bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel socketChannel) throws Exception {
                    ChannelPipeline pipeline = socketChannel.pipeline();
                    // 添加ServerHandler到ChannelPipeline，对workerGroup的SocketChannel（客户端）设置处理器
                    pipeline.addLast(new NettyServerHandler());
                }
            });
            // 设置启动参数，初始化服务器连接队列大小。服务端处理客户端连接请求是顺序处理，一个时间内只能处理一个客户端请求
            // 当有多个客户端同时来请求时，未处理的请求先放入队列中
            bootstrap.option(ChannelOption.SO_BACKLOG, 1024);
            // 绑定端口并启动服务器，bind方法是异步的，sync方法是等待异步操作执行完成，返回ChannelFuture异步对象
            ChannelFuture channelFuture = bootstrap.bind(8888).sync();
            // 等待服务器关闭
            channelFuture.channel().closeFuture().sync();
        } finally {
            // 优雅地关闭boss线程组
            bossGroup.shutdownGracefully();
            // 优雅地关闭worker线程组
            workerGroup.shutdownGracefully();
        }
    }
}

编写服务端处理器handler

public class NettyServerHandler extends ChannelInboundHandlerAdapter {
    /**
     * 当 Channel 已经注册到它的 EventLoop 并且能够处理 I/O 时被调用
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        System.out.println("执行 channelRegistered");
    }

    /**
     * 当 Channel 从它的 EventLoop 注销并且无法处理任何 I/O 时被调
     * 用
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        System.out.println("执行 channelUnregistered");
    }

    /**
     * 当 Channel 处于活动状态时被调用；Channel 已经连接/绑定并且已经就绪
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("执行 channelActive");
    }

    /**
     * 当 Channel 离开活动状态并且不再连接它的远程节点时被调用
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("执行 channelInactive");
    }

    /**
     * 当从 Channel 读取数据时被调用
     *
     * @param ctx
     * @param msg
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

        System.out.println("执行 channelRead");
        // 处理接收到的数据
        ByteBuf byteBuf = (ByteBuf) msg;
        try {
            // 将接收到的字节数据转换为字符串
            String message = byteBuf.toString(CharsetUtil.UTF_8);
            // 打印接收到的消息
            System.out.println("接收到客户端消息为: " + message);
            // 发送响应消息给客户端
            ctx.writeAndFlush(Unpooled.copiedBuffer("我是服务端，我收到你的消息啦~", CharsetUtil.UTF_8));
        } finally {
            // 释放ByteBuf资源
            ReferenceCountUtil.release(byteBuf);
        }

    }

    /**
     * 当 Channel 上的一个读操作完成时被调用，对通道的读取完成的事件或通知。当读取完成可通知发送方或其他的相关方，告诉他们接受方读取完成
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        System.out.println("执行 channelReadComplete");
    }

    /**
     * 当 ChannelnboundHandler.fireUserEventTriggered()方法被调用时被
     * 调用
     *
     * @param ctx
     * @param evt
     * @throws Exception
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        System.out.println("执行 userEventTriggered");
    }

    /**
     * 当 Channel 的可写状态发生改变时被调用。可以通过调用 Channel 的 isWritable()方法
     * * 来检测 Channel 的可写性。与可写性相关的阈值可以通过
     * * Channel.config().setWriteHighWaterMark()和 Channel.config().setWriteLowWaterMark()方法来
     * * 设置
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
        System.out.println("执行 channelWritabilityChanged");
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println("执行 exceptionCaught");
        // 异常处理
        cause.printStackTrace();
        ctx.close();
    }
}

编写客户端启动类

public class NettyClient {
    /**
     * 创建客户端实例并向服务端发送连接请求
     */
    public static void start() {

        // 创建EventLoopGroup，用于处理客户端的I/O操作
        EventLoopGroup groupThread = new NioEventLoopGroup();

        try {
            // 创建Bootstrap实例，客户端启动对象
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(groupThread);
            // 设置服务端Channel类型为NioSocketChannel作为通道实现
            bootstrap.channel(NioSocketChannel.class);
            // 设置客户端处理
            bootstrap.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel socketChannel) throws Exception {
                    socketChannel.pipeline().addLast(new NettyClientHandler());
                }
            });
            // 向服务端发送连接请求
            ChannelFuture channelFuture = bootstrap.connect("localhost", 8888).sync();
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {
            // 优雅地关闭线程
            groupThread.shutdownGracefully();
        }
    }
}

编写客户端处理器handler

public class NettyClientHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        // 连接建立时的处理，发送请求消息给服务器
        ctx.writeAndFlush(Unpooled.copiedBuffer("我是客户端，连接建立成功！", CharsetUtil.UTF_8));
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        // 处理接收到的数据
        ByteBuf byteBuf = (ByteBuf) msg;
        try {
            // 将接收到的字节数据转换为字符串
            String message = byteBuf.toString(CharsetUtil.UTF_8);
            // 打印接收到的消息
            System.out.println("收到服务端响应的消息为: " + message);

            // TODO: 对数据进行业务处理
        } finally {
            // 释放ByteBuf资源
            ReferenceCountUtil.release(byteBuf);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        // 异常处理
        cause.printStackTrace();
        ctx.close();
    }
}