Netty Reactor 线程模型（随记）

港控/mmm° 2023-07-07 05:44 17阅读 0赞

Netty被称为一个高性能、高可扩展性能的异步事件驱动的网络应用程序框架，它极大地简化了TCP和UDP客户端和服务器开发等网络编程。

Netty的Reactor模型中有四个核心概念：

1.   Resources资源（请求和任务） 
2.   Synchronous Event Demultiplexer同步事件复用器
3.  Dispatcher 分配器
4.   Request Handler请求处理器

Reactor 模式设计和实现，我们先了解下它两端的通信方式：

服务端通信序列图如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NkbXhkemI_size_16_color_FFFFFF_t_70][]

客户端通信方式：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NkbXhkemI_size_16_color_FFFFFF_t_70 1][]

Netty 的 IO 线程 NioEventLoop 由于聚合了多路复用器 Selector，可以同时并发处理成百上千个客户端 Channel，由于读写操作都是非阻塞的，这就可以充分提升 IO 线程的运行效率，避免由于频繁 IO 阻塞导致的线程挂起。

Reactor线程模型的两次转变：

Reactor 线程接收请求->分发给线程池处理请求

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NkbXhkemI_size_16_color_FFFFFF_t_70 2][]

mainReactor接收->分发给subReactor读写->具体业务逻辑分发给单独的线程池处理

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NkbXhkemI_size_16_color_FFFFFF_t_70 3][]

上图出自Doug Lea的著名文档《Scalable IO in java》，建议大家可以去看看。

下载地址：[http://gee.cs.oswego.edu/dl/cpjslides/nio.pdf][http_gee.cs.oswego.edu_dl_cpjslides_nio.pdf]

刚开始自己也没太明白这种设计的好处，于是就深入了下源码，看了看设计思想，深有体会。看百遍不如动手写一遍，这样才会有深入体会。

UML图

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NkbXhkemI_size_16_color_FFFFFF_t_70 4][]

代码如下

package com.wywhdgg.netty.example.demo.reactor;
    
    import java.io.IOException;
    import java.net.InetSocketAddress;
    import java.net.ServerSocket;
    import java.nio.ByteBuffer;
    import java.nio.channels.*;
    import java.util.Iterator;
    import java.util.Random;
    import java.util.Set;
    import java.util.concurrent.*;
    import java.util.concurrent.atomic.AtomicInteger;
    
    /***
     *@author dzhbo
     *@date 2019/12/5 23:07
     *@Description:  Reactorc线程模型
     *@version 1.0.0
     */
    public class ReactorServer {
    
        /**
         * 处理业务的线程池
         **/
        private static ExecutorService workThreadPool = Executors.newCachedThreadPool();
    
    
        /**
         * 封装了selector.select()等事件轮询的代码
         */
        abstract class ReactorThread extends Thread {
            /**
             * 通道选择
             **/
            Selector selector;
            /**
             * 任务队列
             **/
            LinkedBlockingQueue<Runnable> taskQueue = new LinkedBlockingQueue<>();
    
            /**
             * Selector监听到有事件后,调用这个方法
             */
            public abstract void handler(SelectableChannel channel) throws Exception;
    
            private ReactorThread() throws IOException {
                selector = Selector.open();
            }
    
            volatile boolean running = false;
    
            @Override
            public void run() {
                // 轮询Selector事件
                while (running) {
                    try {
                        /**执行队列中的任务**/
                        Runnable task;
                        /**非阻塞，如果获取到，就开始处理**/
                        while ((task = taskQueue.poll()) != null) task.run();
                        selector.select(1000);
                        /**获取查询结果  遍历查询结果**/
                        Set<SelectionKey> selected = selector.selectedKeys();
                        Iterator<SelectionKey> iter = selected.iterator();
                        while (iter.hasNext()) {
                            SelectionKey key = iter.next();
                            iter.remove();
                            int readyOps = key.readyOps();
                            // 关注 Read 和 Accept两个事件
                            if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
                                try {
                                    SelectableChannel channel = (SelectableChannel) key.attachment();
                                    channel.configureBlocking(false);
                                    handler(channel);
                                    if (!channel.isOpen()) {
                                        key.cancel(); // 如果关闭了,就取消这个KEY的订阅
                                    }
                                } catch (Exception ex) {
                                    key.cancel(); // 如果有异常,就取消这个KEY的订阅
                                }
                            }
                        }
                        selector.selectNow();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
    
            private SelectionKey register(SelectableChannel channel) throws Exception {
                /***
                  register要以任务提交的形式，让reactor线程去处理的原因：
                  1.因为线程在执行channel注册到selector的过程中，会和调用selector.select()方法的线程争用同一把锁
                 2.select()方法实在eventLoop中通过while循环调用的，争抢的可能性很高，为了让register能更快的执行，就放到同一个线程来处理
                 */
                FutureTask<SelectionKey> futureTask = new FutureTask<>(() -> channel.register(selector, 0, channel));
                taskQueue.add(futureTask);
                return futureTask.get();
            }
    
            private void doStart() {
                if (!running) {
                    running = true;
                    start();
                }
            }
    
        }
    
    
        private ServerSocketChannel serverSocketChannel;
        // 1、创建多个线程 - accept处理reactor线程 (accept线程)
        private ReactorThread[] mainReactorThreads = new ReactorThread[1];
        // 2、创建多个线程 - io处理reactor线程  (I/O线程)
        private ReactorThread[] subReactorThreads = new ReactorThread[8];
    
        /**
         * 初始化线程组
         */
        private void newGroup() throws IOException {
            // 创建IO线程,负责处理客户端连接以后socketChannel的IO读写
            for (int i = 0; i < subReactorThreads.length; i++) {
                subReactorThreads[i] = new ReactorThread() {
                    @Override
                    public void handler(SelectableChannel channel) throws IOException {
                        /**work线程只负责处理IO处理，不处理accept事件*/
                        SocketChannel socketChannel = (SocketChannel) channel;
                        ByteBuffer requestBuffer = ByteBuffer.allocate(1024);
                        while (socketChannel.isOpen() && socketChannel.read(requestBuffer) != -1) {
                            if (requestBuffer.position() > 0) break;
                        }
    
                        if (requestBuffer.position() == 0) {
                            return;
                        }
                        /**写模式切换读
                         初始位置等于限制
                         limit = position;
                         position = 0; //初始位置设置为0
                         mark = -1; //清除标记
                         * **/
                        requestBuffer.flip();
                        byte[] content = new byte[requestBuffer.limit()];
                        requestBuffer.get(content);
                        System.out.println(Thread.currentThread().getName() +
                                "收到数据,IP ADDRESS：" + socketChannel.getRemoteAddress()
                                + " CONTENT:" + new String(content));
                        workThreadPool.submit(() -> {
                            System.out.println("---------------执行业务数据---------------");
                        });
                        // 响应结果 200
                        String response = "HTTP/1.1 200 OK\r\n" +
                                "Content-Length: 11\r\n\r\n" +
                                "Hello World";
                        ByteBuffer buffer = ByteBuffer.wrap(response.getBytes());
                        while (buffer.hasRemaining()) {
                            socketChannel.write(buffer);
                        }
                    }
                };
            }
    
            // 创建mainReactor线程, 只负责处理serverSocketChannel
            for (int i = 0; i < mainReactorThreads.length; i++) {
                mainReactorThreads[i] = new ReactorThread() {
                    AtomicInteger incr = new AtomicInteger(0);
    
                    @Override
                    public void handler(SelectableChannel channel) throws Exception {
                        ServerSocketChannel serverSocketChannel = (ServerSocketChannel) channel;
                        SocketChannel socketChannel = serverSocketChannel.accept();
                        socketChannel.configureBlocking(false);
                        // 收到连接建立的通知之后，分发给I/O线程继续去读取数据
                        int index = incr.getAndIncrement() % subReactorThreads.length;
                        ReactorThread workEventLoop = subReactorThreads[index];
                        workEventLoop.doStart();
                        /**注册channel 信息**/
                        SelectionKey selectionKey = workEventLoop.register(socketChannel);
                        selectionKey.interestOps(SelectionKey.OP_READ);
                        System.out.println(Thread.currentThread().getName() + "收到新连接 : " + socketChannel.getRemoteAddress());
                    }
                };
            }
    
    
        }
    
    
        /**
         * 初始化channel,并且绑定一个eventLoop线程
         *
         * @throws IOException IO异常
         */
        private void initAndRegister() throws Exception {
            // 1、 创建ServerSocketChannel
            serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.configureBlocking(false);
            // 2、 将serverSocketChannel注册到selector
            int index = new Random().nextInt(mainReactorThreads.length);
            mainReactorThreads[index].doStart();
            SelectionKey selectionKey = mainReactorThreads[index].register(serverSocketChannel);
            //注册接收事件
            selectionKey.interestOps(SelectionKey.OP_ACCEPT);
        }
    
    
        /**
         * 绑定端口
         *
         * @throws IOException IO异常
         */
        private void bind() throws IOException {
            //  1、 正式绑定端口，对外服务
            serverSocketChannel.bind(new InetSocketAddress(8080));
            System.out.println("启动完成，端口8080");
        }
    
        public static void main(String[] args) {
            ReactorServer reactorServer = new ReactorServer();
            try {
                //初始化线程组
                reactorServer.newGroup();
                //初始化链接
                reactorServer.initAndRegister();
                //绑定端口号
                reactorServer.bind();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
    
    
    //        Thread-8收到新连接 : /127.0.0.1:51577
    //        Thread-0收到数据,IP ADDRESS：/127.0.0.1:51577 CONTENT:GET / HTTP/1.1
    //        Host: 127.0.0.1:8080
    //        Connection: keep-alive
    //        Cache-Control: max-age=0
    //        Upgrade-Insecure-Requests: 1
    //        User-Agent: Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3945.88 Safari/537.36
    //        Sec-Fetch-User: ?1
    //        Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9
    //Sec-Fetch-Site: cross-site
    //Sec-Fetch-Mode: navigate
    //Accept-Encoding: gzip, deflate, br
    //Accept-Language: zh-CN,zh;q=0.9
    //
    //
    //Thread-8收到新连接 : /127.0.0.1:51578
    //---------------执行业务数据---------------
    //Thread-0收到数据,IP ADDRESS：/127.0.0.1:51577 CONTENT:GET /favicon.ico HTTP/1.1
    //Host: 127.0.0.1:8080
    //Connection: keep-alive
    //Pragma: no-cache
    //Cache-Control: no-cache
    //User-Agent: Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3945.88 Safari/537.36
    //Accept: image/webp,image/apng,image/*,*/*;q=0.8
    //        Sec-Fetch-Site: same-origin
    //        Sec-Fetch-Mode: no-cors
    //        Referer: http://127.0.0.1:8080/
    //        Accept-Encoding: gzip, deflate, br
    //        Accept-Language: zh-CN,zh;q=0.9
    //
    //
    //        ---------------执行业务数据---------------

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NkbXhkemI_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200221105334438.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NkbXhkemI=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NkbXhkemI_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200221105410475.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NkbXhkemI=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NkbXhkemI_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/2020022111103873.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NkbXhkemI=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NkbXhkemI_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20200221111327599.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NkbXhkemI=,size_16,color_FFFFFF,t_70
[http_gee.cs.oswego.edu_dl_cpjslides_nio.pdf]: http://gee.cs.oswego.edu/dl/cpjslides/nio.pdf
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NkbXhkemI_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20200221111845362.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NkbXhkemI=,size_16,color_FFFFFF,t_70