Netty之Reactor模式

一时失言乱红尘 2022-04-25 00:26 255阅读 0赞

## 1. 单线程模型 ##

Reactor单线程模型，指的是所有的IO操作都在同一个NIO线程上面完成，NIO线程的职责如下：

*  1）作为NIO服务端，接收客户端的TCP连接；
 *  2）作为NIO客户端，向服务端发起TCP连接；
 *  3）读取通信对端的请求或者应答消息；
 *  4）向通信对端发送消息请求或者应答消息。

Reactor单线程模型示意图如下所示：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0ppblhZYW4_size_16_color_FFFFFF_t_70][]

**单线程Reactor模式流程：**

*  服务器端的Reactor是一个线程对象，该线程会启动事件循环，并使用Selector(选择器)来实现IO的多路复用。注册一个Acceptor事件处理器到Reactor中，Acceptor事件处理器所关注的事件是ACCEPT事件，这样Reactor会监听客户端向服务器端发起的连接请求事件(ACCEPT事件)。
 *  客户端向服务器端发起一个连接请求，Reactor监听到了该ACCEPT事件的发生并将该ACCEPT事件派发给相应的Acceptor处理器来进行处理。Acceptor处理器通过accept()方法得到与这个客户端对应的连接(SocketChannel)，然后将该连接所关注的READ事件以及对应的READ事件处理器注册到Reactor中，这样一来Reactor就会监听该连接的READ事件了。
 *  当Reactor监听到有读或者写事件发生时，将相关的事件派发给对应的处理器进行处理。比如，读处理器会通过SocketChannel的read()方法读取数据，此时read()操作可以直接读取到数据，而不会堵塞与等待可读的数据到来。
 *  每当处理完所有就绪的感兴趣的I/O事件后，Reactor线程会再次执行select()阻塞等待新的事件就绪并将其分派给对应处理器进行处理。

注意，Reactor的单线程模式的单线程主要是针对于I/O操作而言，也就是所有的I/O的accept()、read()、write()以及connect()操作都在一个线程上完成的。

对于一些小容量应用场景，可以使用单线程模型。但是对于高负载、大并发的应用场景却不合适，主要原因如下：

*  1）一个NIO线程同时处理成百上千的链路，性能上无法支撑，即便NIO线程的CPU负荷达到100%，也无法满足海量消息的编码、解码、读取和发送；
 *  2）当NIO线程负载过重之后，处理速度将变慢，这会导致大量客户端连接超时，超时之后往往会进行重发，这更加重了NIO线程的负载，最终会导致大量消息积压和处理超时，成为系统的性能瓶颈；
 *  3）可靠性问题：一旦NIO线程意外跑飞，或者进入死循环，会导致整个系统通信模块不可用，不能接收和处理外部消息，造成节点故障。

为了解决这些问题，演进出了Reactor多线程模型，下面我们一起学习下Reactor多线程模型

## 2. 多线程模型 ##

Rector多线程模型与单线程模型最大的区别就是有一组NIO线程处理IO操作，它的原理图如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0ppblhZYW4_size_16_color_FFFFFF_t_70 1][]

Reactor多线程模型的特点：

*  1）有专门一个NIO线程-Acceptor线程用于监听服务端，接收客户端的TCP连接请求；
 *  2）网络IO操作-读、写等由一个NIO线程池负责，线程池可以采用标准的JDK线程池实现，它包含一个任务队列和N个可用的线程，由这些NIO线程负责消息的读取、解码、编码和发送；
 *  3）1个NIO线程可以同时处理N条链路，但是1个链路只对应1个NIO线程，防止发生并发操作问题。

在绝大多数场景下，Reactor多线程模型都可以满足性能需求；但是，在极个别特殊场景中，一个NIO线程负责监听和处理所有的客户端连接可能会存在性能问题。例如并发百万客户端连接，或者服务端需要对客户端握手进行安全认证，但是认证本身非常损耗性能。在这类场景下，单独一个Acceptor线程可能会存在性能不足问题，为了解决性能问题，产生了第三种Reactor线程模型-主从Reactor多线程模型

## 3. 主从多线程模型 ##

主从Reactor线程模型的特点是：服务端用于接收客户端连接的不再是个1个单独的NIO线程，而是一个独立的NIO线程池。Acceptor接收到客户端TCP连接请求处理完成后（可能包含接入认证等），将新创建的SocketChannel注册到IO线程池（sub reactor线程池）的某个IO线程上，由它负责SocketChannel的读写和编解码工作。Acceptor线程池仅仅只用于客户端的登陆、握手和安全认证，一旦链路建立成功，就将链路注册到后端subReactor线程池的IO线程上，由IO线程负责后续的IO操作。

它的线程模型如下图所示：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0ppblhZYW4_size_16_color_FFFFFF_t_70 2][]

利用主从NIO线程模型，可以解决1个服务端监听线程无法有效处理所有客户端连接的性能不足问题。因此，在Netty的官方demo中，推荐使用该线程模型。

**它的工作流程总结如下：**

*  ① 注册一个Acceptor事件处理器到mainReactor中，Acceptor事件处理器所关注的事件是ACCEPT事件，这样mainReactor会监听客户端向服务器端发起的连接请求事件(ACCEPT事件)。启动mainReactor的事件循环。
 *  ② 客户端向服务器端发起一个连接请求，mainReactor监听到了该ACCEPT事件并将该ACCEPT事件派发给Acceptor处理器来进行处理。Acceptor处理器通过accept()方法得到与这个客户端对应的连接(SocketChannel)，然后将这个SocketChannel传递给subReactor线程池。
 *  ③ subReactor线程池分配一个subReactor线程给这个SocketChannel，即，将SocketChannel关注的READ事件以及对应的READ事件处理器注册到subReactor线程中。当然你也注册WRITE事件以及WRITE事件处理器到subReactor线程中以完成I/O写操作。Reactor线程池中的每一Reactor线程都会有自己的Selector、线程和分发的循环逻辑。
 *  ④ 当有I/O事件就绪时，相关的subReactor就将事件派发给响应的处理器处理。注意，这里subReactor线程只负责完成I/O的read()操作，在读取到数据后将业务逻辑的处理放入到线程池中完成，若完成业务逻辑后需要返回数据给客户端，则相关的I/O的write操作还是会被提交回subReactor线程来完成。
 *  注意，所以的I/O操作(包括，I/O的accept()、read()、write()以及connect()操作)依旧还是在Reactor线程(mainReactor线程 或 subReactor线程)中完成的。Thread Pool(线程池)仅用来处理非I/O操作的逻辑。

主从Reactor线程模式将“接受客户端的连接请求”和“与该客户端的通信”分在了两个Reactor线程池来完成。mainReactor完成接收客户端连接请求的操作，它不负责与客户端的通信，而是将建立好的连接转交给subReactor线程来完成与客户端的通信，这样一来就不会因为read()数据量太大而导致后面的客户端连接请求得不到即时处理的情况。并且多Reactor线程模式在海量的客户端并发请求的情况下，还可以通过实现subReactor线程池来将海量的连接分发给多个subReactor线程，在多核的操作系统中这能大大提升应用的负载和吞吐量。

事实上，Netty的线程模型并非固定不变，通过在启动辅助类中创建不同的EventLoopGroup实例并通过适当的参数配置，就可以支持上述三种Reactor线程模型。正是因为Netty对Reactor线程模型的支持提供了灵活的定制能力，所以可以满足不同业务场景的性能诉求。

### Netty 与 Reactor模式 ###

Netty的线程模式就是一个实现了Reactor模式的经典模式。

*  结构对应：  
    NioEventLoop ———— Initiation Dispatcher  
    Synchronous EventDemultiplexer ———— Selector  
    Evnet Handler ———— ChannelHandler  
    ConcreteEventHandler ———— 具体的ChannelHandler的实现
 *  模式对应：  
    Netty服务端使用了“多Reactor线程模式”  
    mainReactor ———— bossGroup(NioEventLoopGroup) 中的某个NioEventLoop  
    subReactor ———— workerGroup(NioEventLoopGroup) 中的某个NioEventLoop  
    acceptor ———— ServerBootstrapAcceptor  
    ThreadPool ———— 用户自定义线程池
 *  流程：  
    ① 当服务器程序启动时，会配置ChannelPipeline，ChannelPipeline中是一个ChannelHandler链，所有的事件发生时都会触发Channelhandler中的某个方法，这个事件会在ChannelPipeline中的ChannelHandler链里传播。然后，从bossGroup事件循环池中获取一个NioEventLoop来现实服务端程序绑定本地端口的操作，将对应的ServerSocketChannel注册到该NioEventLoop中的Selector上，并注册ACCEPT事件为ServerSocketChannel所感兴趣的事件。  
    ② NioEventLoop事件循环启动，此时开始监听客户端的连接请求。  
    ③ 当有客户端向服务器端发起连接请求时，NioEventLoop的事件循环监听到该ACCEPT事件，Netty底层会接收这个连接，通过accept()方法得到与这个客户端的连接(SocketChannel)，然后触发ChannelRead事件(即，ChannelHandler中的channelRead方法会得到回调)，该事件会在ChannelPipeline中的ChannelHandler链中执行、传播。  
    ④ ServerBootstrapAcceptor的readChannel方法会该SocketChannel(客户端的连接)注册到workerGroup(NioEventLoopGroup) 中的某个NioEventLoop的Selector上，并注册READ事件为SocketChannel所感兴趣的事件。启动SocketChannel所在NioEventLoop的事件循环，接下来就可以开始客户端和服务器端的通信了。

### Netty代码实例如下： ###

**客户端：**

import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.nio.NioSocketChannel;
    
    import java.net.InetSocketAddress;
    
    public class EchoClient {
    
        private final int port;
        private final String host;
    
        public EchoClient(int port, String host) {
            this.port = port;
            this.host = host;
        }
    
        public void start() throws InterruptedException {
            EventLoopGroup group = new NioEventLoopGroup();/*线程组*/
            try {
                Bootstrap b = new Bootstrap();/*客户端启动必备*/
                b.group(group)
                    .channel(NioSocketChannel.class)/*指明使用NIO进行网络通讯*/
                    .remoteAddress(new InetSocketAddress(host,port))/*配置远程服务器的地址*/
                    .handler(new EchoClientHandler());
                ChannelFuture f = b.connect().sync();/*连接到远程节点，阻塞等待直到连接完成*/
                /*阻塞，直到channel关闭*/
                f.channel().closeFuture().sync();
            } finally {
                group.shutdownGracefully().sync();
            }
        }
    
        public static void main(String[] args) throws InterruptedException {
            new EchoClient(9999,"127.0.0.1").start();
        }
    }

import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    import io.netty.util.CharsetUtil;
    
    public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
    
        /*客户端读取到数据后干什么*/
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg)
                throws Exception {
            System.out.println("Client accetp:"+msg.toString(CharsetUtil.UTF_8));
        }
    
        /*客户端被通知channel活跃（与服务器连接）以后，做事*/
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            //往服务器写数据
            ctx.writeAndFlush(Unpooled.copiedBuffer("Hello,Netty",
                    CharsetUtil.UTF_8));
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
                throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    }

**服务端：**

import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    
    import java.net.InetSocketAddress;
    
    public class EchoServer  {
    
        private final int port;
    
        public EchoServer(int port) {
            this.port = port;
        }
    
        public static void main(String[] args) throws InterruptedException {
            int port = 9999;
            EchoServer echoServer = new EchoServer(port);
            System.out.println("服务器即将启动");
            echoServer.start();
            System.out.println("服务器关闭");
        }
    
        public void start() throws InterruptedException {
            final EchoServerHandler serverHandler = new EchoServerHandler();
            /*线程组*/
            EventLoopGroup group = new NioEventLoopGroup();
            try {
                ServerBootstrap b = new ServerBootstrap();/*服务端启动必备*/
                b.group(group)
                        /*指明使用NIO进行网络通讯*/
                        .channel(NioServerSocketChannel.class)
                        /*指明服务器监听端口*/
                        .localAddress(new InetSocketAddress(port))
                        /*NioServerSocketChannel只负责连接请求，并不处理，需要socketchannel去处理*/
                        /*接收到连接请求，新启一个socket通信，也就是channel，每个channel
                        * 有自己的事件的handler*/
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception {
                                ch.pipeline().addLast(new EchoServerHandler());
                            }
                        });
                ChannelFuture f = b.bind().sync();/*绑定到端口，阻塞等待直到连接完成*/
                /*阻塞，直到channel关闭*/
                f.channel().closeFuture().sync();
            } finally {
                group.shutdownGracefully().sync();
            }
        }
    }

import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelFutureListener;
    import io.netty.channel.ChannelHandler;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import io.netty.util.CharsetUtil;
    
    /*指明我这个handler可以在多个channel之间共享，意味这个实现必须线程安全的。*/
    @ChannelHandler.Sharable
    public class EchoServerHandler extends ChannelInboundHandlerAdapter {
    
        /*** 服务端读取到网络数据后的处理*/
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg)
                throws Exception {
            ByteBuf in = (ByteBuf)msg;/*netty实现的缓冲区*/
            System.out.println("Server Accept:"+in.toString(CharsetUtil.UTF_8));
            ctx.write(in);
        }
    
        /*** 服务端读取完成网络数据后的处理*/
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx)
                throws Exception {
            ctx.writeAndFlush(Unpooled.EMPTY_BUFFER) /*flush掉所有的数据*/
                    .addListener(ChannelFutureListener.CLOSE); /*当flush完成后，关闭连接*/
        }
    
        /*** 发生异常后的处理*/
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
                throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    }

参考文章：

[https://www.cnblogs.com/junzi2099/p/8119465.html][https_www.cnblogs.com_junzi2099_p_8119465.html]

https://www.jianshu.com/p/1ccbc6a348db

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0ppblhZYW4_size_16_color_FFFFFF_t_70]: /images/20220210/12a68a7cc55743dcad0c6d8fbaa7ca75.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0ppblhZYW4_size_16_color_FFFFFF_t_70 1]: /images/20220210/a0cad6630309463aa4281a4df5a11d0a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0ppblhZYW4_size_16_color_FFFFFF_t_70 2]: /images/20220210/4145c882341140caa19291a7de579a3e.png
[https_www.cnblogs.com_junzi2099_p_8119465.html]: https://www.cnblogs.com/junzi2099/p/8119465.html