45 张图深度解析 Netty 架构与原理(四)

简介: 作为一个学 Java 的,如果没有研究过 Netty,那么你对 Java 语言的使用和理解仅仅停留在表面水平,会点 SSH 写几个 MVC,访问数据库和缓存,这些只是初等 Java 程序员干的事。如果你要进阶,想了解 Java 服务器的深层高阶知识,Netty 绝对是一个必须要过的门槛。 接下来我们会学习一个 Netty 系列教程,Netty 系列由「架构与原理」,「源码」,「架构」三部分组成,今天我们先来看看第一部分:Netty 架构与原理初探,大纲如下:

2.3. Netty 的模样

Netty 的设计主要基于主从 Reactor 多线程模式,并做了一定的改进。本节将使用一种渐进式的描述方式展示 Netty 的模样,即先给出 Netty 的简单版本,然后逐渐丰富其细节,直至展示出 Netty 的全貌。

简单版本的 Netty 的模样如下:

116.jpg070

关于这张图,作以下几点说明:

1)BossGroup 线程维护 Selector,ServerSocketChannel 注册到这个 Selector 上,只关注连接建立请求事件(相当于主 Reactor)。

2)当接收到来自客户端的连接建立请求事件的时候,通过 ServerSocketChannel.accept 方法获得对应的 SocketChannel,并封装成 NioSocketChannel 注册到 WorkerGroup 线程中的 Selector,每个 Selector 运行在一个线程中(相当于从 Reactor)。

3)当 WorkerGroup 线程中的 Selector 监听到自己感兴趣的 IO 事件后,就调用 Handler 进行处理。

我们给这简单版的 Netty 添加一些细节:

1177.jpg

关于这张图,作以下几点说明:

1)有两组线程池:BossGroup 和 WorkerGroup,BossGroup 中的线程(可以有多个,图中只画了一个)专门负责和客户端建立连接,WorkerGroup 中的线程专门负责处理连接上的读写。

2)BossGroup 和 WorkerGroup 含有多个不断循环的执行事件处理的线程,每个线程都包含一个 Selector,用于监听注册在其上的 Channel。

3)每个 BossGroup 中的线程循环执行以下三个步骤:

3.1)轮训注册在其上的 ServerSocketChannel 的 accept 事件(OP_ACCEPT 事件)

3.2)处理 accept 事件,与客户端建立连接,生成一个 NioSocketChannel,并将其注册到 WorkerGroup 中某个线程上的 Selector 上

3.3)再去以此循环处理任务队列中的下一个事件

4)每个 WorkerGroup 中的线程循环执行以下三个步骤:

4.1)轮训注册在其上的 NioSocketChannel 的 read/write 事件(OP_READ/OP_WRITE 事件)

4.2)在对应的 NioSocketChannel 上处理 read/write 事件

4.3)再去以此循环处理任务队列中的下一个事件

我们再来看下终极版的 Netty 的模样,如下图所示(图片来源于网络):

118.jpg

关于这张图,作以下几点说明:

1)Netty 抽象出两组线程池:BossGroup 和 WorkerGroup,也可以叫做 BossNioEventLoopGroup 和 WorkerNioEventLoopGroup。每个线程池中都有 NioEventLoop 线程。BossGroup 中的线程专门负责和客户端建立连接,WorkerGroup 中的线程专门负责处理连接上的读写。BossGroup 和 WorkerGroup 的类型都是 NioEventLoopGroup。

2)NioEventLoopGroup 相当于一个事件循环组,这个组中含有多个事件循环,每个事件循环就是一个 NioEventLoop。

3)NioEventLoop 表示一个不断循环的执行事件处理的线程,每个 NioEventLoop 都包含一个 Selector,用于监听注册在其上的 Socket 网络连接(Channel)。

4)NioEventLoopGroup 可以含有多个线程,即可以含有多个 NioEventLoop。

5)每个 BossNioEventLoop 中循环执行以下三个步骤:

5.1)select:轮训注册在其上的 ServerSocketChannel 的 accept 事件(OP_ACCEPT 事件)

5.2)processSelectedKeys:处理 accept 事件,与客户端建立连接,生成一个 NioSocketChannel,并将其注册到某个 WorkerNioEventLoop 上的 Selector 上

5.3)runAllTasks:再去以此循环处理任务队列中的其他任务

6)每个 WorkerNioEventLoop 中循环执行以下三个步骤:

6.1)select:轮训注册在其上的 NioSocketChannel 的 read/write 事件(OP_READ/OP_WRITE 事件)

6.2)processSelectedKeys:在对应的 NioSocketChannel 上处理 read/write 事件

6.3)runAllTasks:再去以此循环处理任务队列中的其他任务

7)在以上两个processSelectedKeys步骤中,会使用 Pipeline(管道),Pipeline 中引用了 Channel,即通过 Pipeline 可以获取到对应的 Channel,Pipeline 中维护了很多的处理器(拦截处理器、过滤处理器、自定义处理器等)。这里暂时不详细展开讲解 Pipeline。

2.4. 基于 Netty 的 TCP Server/Client 案例

下面我们写点代码来加深理解 Netty 的模样。下面两段代码分别是基于 Netty 的 TCP Server 和 TCP Client。

服务端代码为:

/**
 * 需要的依赖:
 * <dependency>
 * <groupId>io.netty</groupId>
 * <artifactId>netty-all</artifactId>
 * <version>4.1.52.Final</version>
 * </dependency>
 */
public static void main(String[] args) throws InterruptedException {
    // 创建 BossGroup 和 WorkerGroup
    // 1. bossGroup 只处理连接请求
    // 2. 业务处理由 workerGroup 来完成
    EventLoopGroup bossGroup = new NioEventLoopGroup();
    EventLoopGroup workerGroup = new NioEventLoopGroup();
    try {
        // 创建服务器端的启动对象
        ServerBootstrap bootstrap = new ServerBootstrap();
        // 配置参数
        bootstrap
                // 设置线程组
                .group(bossGroup, workerGroup)
                // 说明服务器端通道的实现类(便于 Netty 做反射处理)
                .channel(NioServerSocketChannel.class)
                // 设置等待连接的队列的容量(当客户端连接请求速率大
             // 于 NioServerSocketChannel 接收速率的时候,会使用
                // 该队列做缓冲)
                // option()方法用于给服务端的 ServerSocketChannel
                // 添加配置
                .option(ChannelOption.SO_BACKLOG, 128)
                // 设置连接保活
                // childOption()方法用于给服务端 ServerSocketChannel
                // 接收到的 SocketChannel 添加配置
                .childOption(ChannelOption.SO_KEEPALIVE, true)
                // handler()方法用于给 BossGroup 设置业务处理器
                // childHandler()方法用于给 WorkerGroup 设置业务处理器
                .childHandler(
                        // 创建一个通道初始化对象
                        new ChannelInitializer<SocketChannel>() {
                            // 向 Pipeline 添加业务处理器
                            @Override
                            protected void initChannel(
                                    SocketChannel socketChannel
                            ) throws Exception {
                                socketChannel.pipeline().addLast(
                                        new NettyServerHandler()
                                );
                                // 可以继续调用 socketChannel.pipeline().addLast()
                                // 添加更多 Handler
                            }
                        }
                );
        System.out.println("server is ready...");
        // 绑定端口,启动服务器,生成一个 channelFuture 对象,
        // ChannelFuture 涉及到 Netty 的异步模型,后面展开讲
        ChannelFuture channelFuture = bootstrap.bind(8080).sync();
        // 对通道关闭进行监听
        channelFuture.channel().closeFuture().sync();
    } finally {
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
    }
}
/**
 * 自定义一个 Handler,需要继承 Netty 规定好的某个 HandlerAdapter(规范)
 * InboundHandler 用于处理数据流入本端(服务端)的 IO 事件
 * InboundHandler 用于处理数据流出本端(服务端)的 IO 事件
 */
static class NettyServerHandler extends ChannelInboundHandlerAdapter {
    /**
     * 当通道有数据可读时执行
     *
     * @param ctx 上下文对象,可以从中取得相关联的 Pipeline、Channel、客户端地址等
     * @param msg 客户端发送的数据
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg)
            throws Exception {
        // 接收客户端发来的数据
        System.out.println("client address: "
                + ctx.channel().remoteAddress());
        // ByteBuf 是 Netty 提供的类,比 NIO 的 ByteBuffer 性能更高
        ByteBuf byteBuf = (ByteBuf) msg;
        System.out.println("data from client: "
                + byteBuf.toString(CharsetUtil.UTF_8));
    }
    /**
     * 数据读取完毕后执行
     *
     * @param ctx 上下文对象
     * @throws Exception
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx)
            throws Exception {
        // 发送响应给客户端
        ctx.writeAndFlush(
                // Unpooled 类是 Netty 提供的专门操作缓冲区的工具
                // 类,copiedBuffer 方法返回的 ByteBuf 对象类似于
                // NIO 中的 ByteBuffer,但性能更高
                Unpooled.copiedBuffer(
                        "hello client! i have got your data.",
                        CharsetUtil.UTF_8
                )
        );
    }
    /**
     * 发生异常时执行
     *
     * @param ctx   上下文对象
     * @param cause 异常对象
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
            throws Exception {
        // 关闭与客户端的 Socket 连接
        ctx.channel().close();
    }
}

客户端端代码为:

/**
 * 需要的依赖:
 * <dependency>
 * <groupId>io.netty</groupId>
 * <artifactId>netty-all</artifactId>
 * <version>4.1.52.Final</version>
 * </dependency>
 */
public static void main(String[] args) throws InterruptedException {
    // 客户端只需要一个事件循环组,可以看做 BossGroup
    EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
    try {
        // 创建客户端的启动对象
        Bootstrap bootstrap = new Bootstrap();
        // 配置参数
        bootstrap
                // 设置线程组
                .group(eventLoopGroup)
                // 说明客户端通道的实现类(便于 Netty 做反射处理)
                .channel(NioSocketChannel.class)
                // handler()方法用于给 BossGroup 设置业务处理器
                .handler(
                        // 创建一个通道初始化对象
                        new ChannelInitializer<SocketChannel>() {
                            // 向 Pipeline 添加业务处理器
                            @Override
                            protected void initChannel(
                                    SocketChannel socketChannel
                            ) throws Exception {
                                socketChannel.pipeline().addLast(
                                        new NettyClientHandler()
                                );
                                // 可以继续调用 socketChannel.pipeline().addLast()
                                // 添加更多 Handler
                            }
                        }
                );
        System.out.println("client is ready...");
        // 启动客户端去连接服务器端,ChannelFuture 涉及到 Netty 的异步模型,后面展开讲
        ChannelFuture channelFuture = bootstrap.connect(
                "127.0.0.1",
                8080).sync();
        // 对通道关闭进行监听
        channelFuture.channel().closeFuture().sync();
    } finally {
        eventLoopGroup.shutdownGracefully();
    }
}
/**
 * 自定义一个 Handler,需要继承 Netty 规定好的某个 HandlerAdapter(规范)
 * InboundHandler 用于处理数据流入本端(客户端)的 IO 事件
 * InboundHandler 用于处理数据流出本端(客户端)的 IO 事件
 */
static class NettyClientHandler extends ChannelInboundHandlerAdapter {
    /**
     * 通道就绪时执行
     *
     * @param ctx 上下文对象
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx)
            throws Exception {
        // 向服务器发送数据
        ctx.writeAndFlush(
                // Unpooled 类是 Netty 提供的专门操作缓冲区的工具
                // 类,copiedBuffer 方法返回的 ByteBuf 对象类似于
                // NIO 中的 ByteBuffer,但性能更高
                Unpooled.copiedBuffer(
                        "hello server!",
                        CharsetUtil.UTF_8
                )
        );
    }
    /**
     * 当通道有数据可读时执行
     *
     * @param ctx 上下文对象
     * @param msg 服务器端发送的数据
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg)
            throws Exception {
        // 接收服务器端发来的数据
        System.out.println("server address: "
                + ctx.channel().remoteAddress());
        // ByteBuf 是 Netty 提供的类,比 NIO 的 ByteBuffer 性能更高
        ByteBuf byteBuf = (ByteBuf) msg;
        System.out.println("data from server: "
                + byteBuf.toString(CharsetUtil.UTF_8));
    }
    /**
     * 发生异常时执行
     *
     * @param ctx   上下文对象
     * @param cause 异常对象
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
            throws Exception {
        // 关闭与服务器端的 Socket 连接
        ctx.channel().close();
    }
}

什么?你觉得使用 Netty 编程难度和工作量更大了?不会吧不会吧,你要知道,你通过这么两段简短的代码得到了一个基于主从 Reactor 多线程模式的服务器,一个高吞吐量和并发量的服务器,一个异步处理服务器……你还要怎样?

对上面的两段代码,作以下简单说明:

1)Bootstrap 和 ServerBootstrap 分别是客户端和服务器端的引导类,一个 Netty 应用程序通常由一个引导类开始,主要是用来配置整个 Netty 程序、设置业务处理类(Handler)、绑定端口、发起连接等。

2)客户端创建一个 NioSocketChannel 作为客户端通道,去连接服务器。

3)服务端首先创建一个 NioServerSocketChannel 作为服务器端通道,每当接收一个客户端连接就产生一个 NioSocketChannel 应对该客户端。

4)使用 Channel 构建网络 IO 程序的时候,不同的协议、不同的阻塞类型和 Netty 中不同的 Channel 对应,常用的 Channel 有:

  • NioSocketChannel:非阻塞的 TCP 客户端 Channel(本案例的客户端使用的 Channel)
  • NioServerSocketChannel:非阻塞的 TCP 服务器端 Channel(本案例的服务器端使用的 Channel)
  • NioDatagramChannel:非阻塞的 UDP Channel
  • NioSctpChannel:非阻塞的 SCTP 客户端 Channel
  • NioSctpServerChannel:非阻塞的 SCTP 服务器端 Channel
    ......

启动服务端和客户端代码,调试以上的服务端代码,发现:

1)默认情况下 BossGroup 和 WorkerGroup 都包含 16 个线程(NioEventLoop),这是因为我的 PC 是 8 核的 NioEventLoop 的数量=coreNum*2。这 16 个线程相当于主 Reactor。

119.jpg120.jpg121.jpg

其实创建 BossGroup 和 WorkerGroup 的时候可以指定 NioEventLoop 数量,如下:

EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup(16);

这样就能更好地分配线程资源。

2)每一个 NioEventLoop 包含如下的属性(比如自己的 Selector、任务队列、执行器等):

122.jpg

3)将代码断在服务端的 NettyServerHandler.channelRead 上:

123.jpg

可以看到 ctx 中包含的属性如下:

124.jpg

可以看到:

  • 当前 ChannelHandlerContext ctx 是位于 ChannelHandlerContext 责任链中的一环,可以看到其 next、prev 属性
  • 当前 ChannelHandlerContext ctx 包含一个 Handler
  • 当前 ChannelHandlerContext ctx 包含一个 Pipeline
  • Pipeline 本质上是一个双向循环列表,可以看到其 tail、head 属性
  • Pipeline 中包含一个 Channel,Channel 中又包含了该 Pipeline,两者互相引用
    ……

从下一节开始,我将深入剖析以上两段代码,向读者展示 Netty 的更多细节。

相关文章
|
机器学习/深度学习 文字识别 监控
安全监控系统:技术架构与应用解析
该系统采用模块化设计,集成了行为识别、视频监控、人脸识别、危险区域检测、异常事件检测、日志追溯及消息推送等功能,并可选配OCR识别模块。基于深度学习与开源技术栈(如TensorFlow、OpenCV),系统具备高精度、低延迟特点,支持实时分析儿童行为、监测危险区域、识别异常事件,并将结果推送给教师或家长。同时兼容主流硬件,支持本地化推理与分布式处理,确保可靠性与扩展性,为幼儿园安全管理提供全面解决方案。
881 3
|
安全 算法 网络协议
解析:HTTPS通过SSL/TLS证书加密的原理与逻辑
HTTPS通过SSL/TLS证书加密,结合对称与非对称加密及数字证书验证实现安全通信。首先,服务器发送含公钥的数字证书,客户端验证其合法性后生成随机数并用公钥加密发送给服务器,双方据此生成相同的对称密钥。后续通信使用对称加密确保高效性和安全性。同时,数字证书验证服务器身份,防止中间人攻击;哈希算法和数字签名确保数据完整性,防止篡改。整个流程保障了身份认证、数据加密和完整性保护。
|
机器学习/深度学习 数据可视化 PyTorch
深入解析图神经网络注意力机制:数学原理与可视化实现
本文深入解析了图神经网络(GNNs)中自注意力机制的内部运作原理,通过可视化和数学推导揭示其工作机制。文章采用“位置-转移图”概念框架,并使用NumPy实现代码示例,逐步拆解自注意力层的计算过程。文中详细展示了从节点特征矩阵、邻接矩阵到生成注意力权重的具体步骤,并通过四个类(GAL1至GAL4)模拟了整个计算流程。最终,结合实际PyTorch Geometric库中的代码,对比分析了核心逻辑,为理解GNN自注意力机制提供了清晰的学习路径。
1090 7
深入解析图神经网络注意力机制:数学原理与可视化实现
|
机器学习/深度学习 缓存 自然语言处理
深入解析Tiktokenizer:大语言模型中核心分词技术的原理与架构
Tiktokenizer 是一款现代分词工具,旨在高效、智能地将文本转换为机器可处理的离散单元(token)。它不仅超越了传统的空格分割和正则表达式匹配方法,还结合了上下文感知能力,适应复杂语言结构。Tiktokenizer 的核心特性包括自适应 token 分割、高效编码能力和出色的可扩展性,使其适用于从聊天机器人到大规模文本分析等多种应用场景。通过模块化设计,Tiktokenizer 确保了代码的可重用性和维护性,并在分词精度、处理效率和灵活性方面表现出色。此外,它支持多语言处理、表情符号识别和领域特定文本处理,能够应对各种复杂的文本输入需求。
1787 6
深入解析Tiktokenizer:大语言模型中核心分词技术的原理与架构
|
存储 机器学习/深度学习 应用服务中间件
阿里云服务器架构解析:从X86到高性能计算、异构计算等不同架构性能、适用场景及选择参考
当我们准备选购阿里云服务器时,阿里云提供了X86计算、ARM计算、GPU/FPGA/ASIC、弹性裸金属服务器以及高性能计算等多种架构,每种架构都有其独特的特点和适用场景。本文将详细解析这些架构的区别,探讨它们的主要特点和适用场景,并为用户提供选择云服务器架构的全面指南。
1491 18
|
传感器 人工智能 监控
反向寻车系统怎么做?基本原理与系统组成解析
本文通过反向寻车系统的核心组成部分与技术分析,阐述反向寻车系统的工作原理,适用于适用于商场停车场、医院停车场及火车站停车场等。如需获取智慧停车场反向寻车技术方案前往文章最下方获取,如有项目合作及技术交流欢迎私信作者。
1253 2
|
算法 前端开发 定位技术
地铁站内导航系统解决方案:技术架构与核心功能设计解析
本文旨在分享一套地铁站内导航系统技术方案,通过蓝牙Beacon技术与AI算法的结合,解决传统导航定位不准确、路径规划不合理等问题,提升乘客出行体验,同时为地铁运营商提供数据支持与增值服务。 如需获取校地铁站内智能导航系统方案文档可前往文章最下方获取,如有项目合作及技术交流欢迎私信我们哦~
1499 1
|
开发框架 监控 JavaScript
解锁鸿蒙装饰器:应用、原理与优势全解析
ArkTS提供了多维度的状态管理机制。在UI开发框架中,与UI相关联的数据可以在组件内使用,也可以在不同组件层级间传递,比如父子组件之间、爷孙组件之间,还可以在应用全局范围内传递或跨设备传递。
597 2
|
负载均衡 JavaScript 前端开发
分片上传技术全解析:原理、优势与应用(含简单实现源码)
分片上传通过将大文件分割成多个小的片段或块,然后并行或顺序地上传这些片段,从而提高上传效率和可靠性,特别适用于大文件的上传场景,尤其是在网络环境不佳时,分片上传能有效提高上传体验。 博客不应该只有代码和解决方案,重点应该在于给出解决方案的同时分享思维模式,只有思维才能可持续地解决问题,只有思维才是真正值得学习和分享的核心要素。如果这篇博客能给您带来一点帮助,麻烦您点个赞支持一下,还可以收藏起来以备不时之需,有疑问和错误欢迎在评论区指出~
|
设计模式 存储 安全
【23种设计模式·全精解析 | 创建型模式篇】5种创建型模式的结构概述、实现、优缺点、扩展、使用场景、源码解析
结构型模式描述如何将类或对象按某种布局组成更大的结构。它分为类结构型模式和对象结构型模式,前者采用继承机制来组织接口和类,后者釆用组合或聚合来组合对象。由于组合关系或聚合关系比继承关系耦合度低,满足“合成复用原则”,所以对象结构型模式比类结构型模式具有更大的灵活性。 结构型模式分为以下 7 种: • 代理模式 • 适配器模式 • 装饰者模式 • 桥接模式 • 外观模式 • 组合模式 • 享元模式
1085 140
【23种设计模式·全精解析 | 创建型模式篇】5种创建型模式的结构概述、实现、优缺点、扩展、使用场景、源码解析

推荐镜像

更多
  • DNS