Netty剖析之Netty线程模型

素颜马尾好姑娘i 2023-07-03 06:06 16阅读 0赞

### 线程模型基本介绍 ###

不同的线程模式，对程序的性能有很大影响，为了搞清Netty线程模式，下面我们来系统的讲解下各个线程模式。

目前存在的线程模型有：

1.  传统阻塞IO服务模型
2.  Reactor 模式，根据 Reactor 的数量和处理资源池线程的数量不同，有3种典型的实现：  
    2.1 单Reactor单线程  
    2.2 单Reactor多线程  
    2.3 主从Reactor多线程
3.  Netty线程模式(Netty主要基于主从Reactor多线程模型做了一定的改进，其中主从Reactor多线程模型有多个 Reactor)

### 传统阻塞IO服务模型 ###

示意图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n_size_16_color_FFFFFF_t_70]

模型特点：

1.  采用阻塞IO模式获取输入的数据
2.  每个连接都需要独立的线程完成数据的输入，业务处理，数据返回

问题分析：

1.  当并发数很大，就会创建大量的线程，占用很大系统资源
2.  连接创建后，如果当前线程暂时没有数据可读，该线程会阻塞在read操作，造成线程资源浪费

### Reactor模式 ###

针对传统阻塞IO服务模型的2个缺点，产生了如下解决方案：

1.  基于IO复用模型：多个连接共用一个阻塞对象，应用程序只需要在一个阻塞对象等待，无需阻塞等待所有连接。当某个连接有新的数据可以处理时，操作系统通知应用程序，线程从阻塞状态返回，开始进行业务处理
2.  基于线程池复用线程资源：不必再为每个连接创建线程，将连接完成后的业务处理任务分配给线程进行处理，一个线程可以处理多个连接的业务。

Reactor通常有如下几种称呼：

1.  反应器模式
2.  分发者模式(Dispatcher)
3.  通知者模式(notifier)

**IO复用结合线程池，就是Reactor模式基本设计思想**，如下图所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n_size_16_color_FFFFFF_t_70 1]  
Reactor模型说明：

*  Reactor 模式，通过一个或多个输入同时传递给服务处理器的模式(基于事件驱动)
 *  服务器端程序处理传入的多个请求,并将它们同步分派到相应的处理线程， 因此Reactor模式也叫 Dispatcher模式
 *  Reactor 模式使用IO复用监听事件, 收到事件后，分发给某个线程(进程), 这点就是网络服务器高并发处理关键

Reactor核心组成部分：

1.  Reactor：Reactor 在一个单独的线程中运行，负责监听和分发事件，分发给适当的处理程序来对IO事件做出反应。 它就像公司的电话接线员，它接听来自客户的电话并将线路转移到适当的联系人；
2.  Handlers：处理程序执行IO事件要完成的实际事件，类似于客户想要与之交谈的公司中的实际官员。Reactor 通过调度适当的处理程序来响应IO事件，处理程序执行非阻塞操作；

根据Reactor的数量和处理资源池线程的数量不同，有3种典型的实现：

1.  单Reactor单线程
2.  单Reactor多线程
3.  主从Reactor多线程

下面我将分别详细来介绍

### 单Reactor单线程 ###

示意图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n_size_16_color_FFFFFF_t_70 2]

工作原理：

1.  Select 是前面 I/O 复用模型介绍的标准网络编程 API，可以实现应用程序通过一个阻塞对象监听多路连接请求
2.  Reactor 对象通过 Select 监控客户端请求事件，收到事件后通过 Dispatch 进行分发
3.  如果是建立连接请求事件，则由 Acceptor 通过 Accept 处理连接请求，然后创建一个Handler对象处理连接完成后的后续业务处理
4.  如果不是建立连接事件，则Reactor会分发调用连接对应的Handler来响应
5.  Handler会完成 Read→业务处理→Send 的完整业务流程

优点:

*  模型简单，没有多线程、进程通信、竞争的问题，全部都在一个线程中完成

缺点：

*  性能问题，只有一个线程，无法完全发挥多核 CPU 的性能。Handler 在处理某个连接上的业务时，整个进程无法处理其他连接事件，很容易导致性能瓶颈
 *  可靠性问题，线程意外终止，或者进入死循环，会导致整个系统通信模块不可用，不能接收和处理外部消息，造成节点故障

使用场景：客户端的数量有限，业务处理非常快速，比如 Redis在业务处理的时间复杂度 O(1) 的情况

### 单Reactor多线程 ###

示意图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n_size_16_color_FFFFFF_t_70 3]  
工作原理：

1.  Reactor 对象通过select 监控客户端请求事件, 收到事件后，通过dispatch进行分发
2.  如果建立连接请求, 则右Acceptor通过accept处理连接请求, 然后创建一个Handler对象处理完成连接后的各种事件
3.  如果不是连接请求，则由reactor分发调用连接对应的handler来处理
4.  handler只负责响应事件，不做具体的业务处理, 通过read读取数据后，会分发给后面的worker线程池的某个线程处理业务
5.  worker线程池会分配独立线程完成真正的业务，并将结果返回给handler
6.  handler收到响应后，通过send将结果返回给client

优点：

*  可以充分的利用多核cpu 的处理能力

缺点：

*  多线程数据共享和访问比较复杂， reactor 处理所有的事件的监听和响应，在单线程运行， 在高并发场景容易出现性能瓶颈

### 主从Reactor多线程 ###

示意图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n_size_16_color_FFFFFF_t_70 4]  
工作原理：

1.  Reactor主线程 MainReactor 对象通过select 监听连接事件, 收到事件后，通过Acceptor 处理连接事件
2.  当 Acceptor处理连接事件后，MainReactor 将连接分配给SubReactor
3.  subreactor 将连接加入到连接队列进行监听,并创建handler进行各种事件处理
4.  当有新事件发生时， subreactor 就会调用对应的handler处理
5.  handler 通过read 读取数据，分发给后面的worker 线程处理
6.  worker 线程池分配独立的worker 线程进行业务处理，并返回结果
7.  handler 收到响应的结果后，再通过send 将结果返回给client
8.  Reactor 主线程可以对应多个Reactor 子线程, 即MainRecator 可以关联多个SubReactor

优点：

*  父线程与子线程的数据交互简单职责明确，父线程只需要接收新连接，子线程完成后续的业务处理
 *  父线程与子线程的数据交互简单，Reactor 主线程只需要把新连接传给子线程，子线程无需返回数据

缺点：

*  编程复杂度较高

### Reactor模式优点 ###

1.  响应快，不必为单个同步时间所阻塞，虽然 Reactor 本身依然是同步的
2.  可以最大程度的避免复杂的多线程及同步问题，并且避免了多线程/进程的切换开销
3.  扩展性好，可以方便的通过增加 Reactor 实例个数来充分利用 CPU 资源
4.  复用性好，Reactor 模型本身与具体事件处理逻辑无关，具有很高的复用性

### Netty模型 ###

示意图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n_size_16_color_FFFFFF_t_70 5]  
工作原理：

1.  Netty抽象出两组线程池 BossGroup 专门负责接收客户端的连接, WorkerGroup 专门负责网络的读写
2.  BossGroup 和 WorkerGroup 类型都是 NioEventLoopGroup
3.  NioEventLoopGroup 相当于一个事件循环组, 这个组中含有多个事件循环 ，每一个事件循环是NioEventLoop
4.  NioEventLoop 表示一个不断循环的执行处理任务的线程， 每个NioEventLoop 都有一个selector , 用于监听绑定在其上的socket的网络通讯
5.  NioEventLoopGroup 可以有多个线程, 即可以含有多个NioEventLoop
6.  每个Boss NioEventLoop 循环执行的步骤有3步：  
    6.1 轮询accept 事件  
    6.2 处理accept 事件 , 与client建立连接 , 生成NioScocketChannel , 并将其注册到某个worker NIOEventLoop 上的 selector  
    6.3 处理任务队列的任务，即 runAllTasks
7.  每个Worker NIOEventLoop 循环执行的步骤：  
    7.1 轮询read, write 事件  
    7.2 处理i/o事件， 即read , write 事件，在对应NioScocketChannel处理  
    7.3 处理任务队列的任务，即 runAllTasks
8.  每个Worker NIOEventLoop 处理业务时，会使用pipeline(管道), pipeline 中包含了 channel , 即通过pipeline 可以获取到对应通道, 管道中维护了很多的处理器

### Netty入门案例 ###

需求：使用Netty实现客户端可以向服务端发送消息，服务端并可以回复消息给客户端

Netty服务端：

public class NettyServer {
    
        public static void main(String[] args) throws Exception{
            /**
             * 分别创建bossGroup和workerGroup
             *
             * NioEventLoopGroup可以指定子线程数量，默认是cpu的核数*2
             * new NioEventLoopGroup(1)则只会创建一个子线程
             */
            NioEventLoopGroup bossGroup = new NioEventLoopGroup(); // bossGroup用来处理与客户端的连接请求
            NioEventLoopGroup workerGroup = new NioEventLoopGroup(); // workerGroup用来处理客户端业务
    
            try {
                // 创建服务端启动对象，服务端是ServerBootstrap，客户端是Bootstrap
                ServerBootstrap serverBootstrap = new ServerBootstrap();
                // 配置启动对象
                serverBootstrap.group(bossGroup, workerGroup) // 设置事件循环组
                        .channel(NioServerSocketChannel.class) // 设置通道实现类
                        .option(ChannelOption.SO_BACKLOG, 128) // 设置线程队列的连接个数
                        .childOption(ChannelOption.SO_KEEPALIVE, true) // 设置保持活动连接状态
                        .childHandler(new ChannelInitializer<SocketChannel>() { // 创建一个通道测试对象
                            @Override
                            protected void initChannel(SocketChannel socketChannel) throws Exception {
                                // 给pipeline设置处理器
                                socketChannel.pipeline().addLast(new NettyServerHandler());
                            }
                        }); // 给线程组设置对应的管道处理器
    
                System.out.println("netty server is ready....");
    
                // 绑定端口
                ChannelFuture channelFuture = serverBootstrap.bind(8899).sync();
                // 关闭通道监听
                channelFuture.channel().closeFuture().sync();
            } finally {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    }

自定义服务端通道处理handler：

public class NettyServerHandler extends ChannelInboundHandlerAdapter {
    
        /**
         * 数据读取处理方法
         *
         * @param ctx 上下文对象
         * @param msg 客户端发送的数据
         * @throws Exception
         */
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    
            System.out.println("server channelRead thread ：" + Thread.currentThread().getName());
    
            Channel channel = ctx.channel();
    
    
            // 此处的ByteBuf是Nttey提供的，不是NIO中的ByteBuffer
            ByteBuf byteBuf = (ByteBuf) msg;
            System.out.println("收到来自[" + channel.remoteAddress() + "]的消息：" + byteBuf.toString(CharsetUtil.UTF_8));
        }
    
        /**
         * 数据读取完成处理方法
         * channelRead方法执行完后才会执行此方法
         *
         * @param ctx
         * @throws Exception
         */
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            // 数据读取完成之后，我们向客户端发送一个消息
            String msg = "hello client，msg is read！";
            // 消息编码
            ByteBuf byteBuf = Unpooled.copiedBuffer(msg, CharsetUtil.UTF_8);
            // 将数据写入到缓存，并刷新 ，同时执行了两个操作 write + flush
            ctx.writeAndFlush(byteBuf);
        }
    
        /**
         * 异常处理方法
         *
         * @param ctx
         * @param cause
         * @throws Exception
         */
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            ctx.close();
        }
    }

Netty客户端：

public class NettyClient {
    
        public static void main(String[] args) throws Exception {
            // 客户端只需要创建一个事件循环组
            NioEventLoopGroup group = new NioEventLoopGroup();
    
            try {
                // 创建客户端启动对象，客户端是Bootstrap，服务端是ServerBootstrap
                Bootstrap bootstrap = new Bootstrap();
                // 配置启动对象
                bootstrap.group(group) // 设置事件循环组
                        .channel(NioSocketChannel.class) // 设置通道实现类
                        .handler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel socketChannel) throws Exception {
                                // 给pipeline设置处理器
                                socketChannel.pipeline().addLast(new NeetyClientHandler());
                            }
                        });
    
                System.out.println("netty client is ready....");
    
                // 连接服务端
                ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8899).sync();
                // 关闭通道监听
                channelFuture.channel().closeFuture().sync();
            } finally {
                group.shutdownGracefully();
            }
        }
    }

自定义客户端通道处理handler：

public class NeetyClientHandler extends ChannelInboundHandlerAdapter {
    
        /**
         * 通道就绪处理方法
         *
         * @param ctx
         * @throws Exception
         */
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            // 通道就绪后，我们向服务端发送一个消息
            String msg = "hello server！";
            // 消息编码
            ByteBuf byteBuf = Unpooled.copiedBuffer(msg, CharsetUtil.UTF_8);
            // 将数据写入到缓存，并刷新 ，同时执行了两个操作 write + flush
            ctx.writeAndFlush(byteBuf);
        }
    
        /**
         * 数据读取处理方法
         *
         * @param ctx 上下文对象
         * @param msg 客户端发送的数据
         * @throws Exception
         */
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    
            Channel channel = ctx.channel();
    
            // 此处的ByteBuf是Nttey提供的，不是NIO中的ByteBuffer
            ByteBuf byteBuf = (ByteBuf) msg;
            System.out.println("收到来自[" + channel.remoteAddress() + "]的消息：" + byteBuf.toString(CharsetUtil.UTF_8));
        }
    
        /**
         * 异常处理方法
         *
         * @param ctx
         * @param cause
         * @throws Exception
         */
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            ctx.close();
        }
    }

测试：

netty server is ready....
    收到来自[/127.0.0.1:7007]的消息：hello server！

netty client is ready....
    收到来自[/127.0.0.1:8899]的消息：hello client，msg is read！

客户端与服务端通信成功！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200202215323342.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200202215334699.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200202215359798.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20200202215408344.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20200202215427770.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/20200202215434413.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n,size_16,color_FFFFFF,t_70