Netty Reactor模式

Love The Way You Lie 2023-06-15 06:38 17阅读 0赞

### 文章目录 ###

*  Netty Reactor反应器模式
 *   *  基本介绍
     *   *  简介
     *  单线程Reactor反应器模式
     *   *  单线程Reactor反应器
         *  方法
         *  一个Reactoor反应器版本的EchoServer实践案例
         *   *  EchoClient
             *  EchoHandler
             *  EchoServerReactor
             *  单线程Reactor缺点
         *  多线程的Reactor反应器模式
         *   *  多线程reactor反应器演进
             *  多线程Reactor反应器的实践
             *  MultiThreadEchoServerReactor
             *  MultiThreadEchoHandler

# Netty Reactor反应器模式 #

## 基本介绍 ##

> Netty整体框架，就是基于反应器模式

### 简介 ###

> 反应器模式由Reactor反应器线程，Handlers处理器两大角色组成

*  Reactor反应器线程的职责： 负责相应IO事件，并且分发到Handler处理器
 *  Handlers处理器的职责： 非阻塞的执行业务处理逻辑

## 单线程Reactor反应器模式 ##

> 总体来说，Reactor反应器模式类似于事件驱动

事件驱动模式

> 当有事件触发时，事件源会将时间dispatch分发到handler处理器进行事件处理，反应器模式中的反应器角色，类似于事件驱动模式中dispatcher事件分发器角色

reactor反应模式

> 有Reactor反应器和Handler处理器两个重要组件

*  Reactor反应器： 负责查询IO事件，当检测到一个IO事件，将其发送给相应的Handler处理器去处理，这里IO事件，就是NIO中选择器监控的通道IO事件
 *  Handler处理器： 与IO事件(选择器)绑定，负责IO事件的处理，完成真正的连接建立，通道的读取，处理业务逻辑，负责将结果写出通道等

### 单线程Reactor反应器 ###

> Reactor反应器和Handler处理器处于一个线程

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-yuVFF23y-1574235792516)(63C9036D311A4D499D84EAF1ABCB1542)\]

### 方法 ###

> `void attach(Object o)` 可以将任何的POJO对象，添加到SelectionKey实例, 相当于附件属性的`setter`方法

> 单线程模式中，需要将Handler处理器实例，作为附件添加到SelectionKey实例

> `object attachment()` 取出通过attach(Object o) 添加的附件，相当于`getter`方法

这两个方法配套使用

> 需要进行attach和attachment结合使用，在选择键注册完成之后，调用attach方法， 将handler处理器绑定到选择键， 当事件发生时，调用attachment方法，可以从选择键取出handler处理器，将事件分发到handler处理器中，完成业务处理

### 一个Reactoor反应器版本的EchoServer实践案例 ###

#### EchoClient ####

import util.Dateutil;
    import util.Logger;
    import util.Print;
    
    import java.io.IOException;
    import java.net.InetSocketAddress;
    import java.nio.ByteBuffer;
    import java.nio.channels.SelectionKey;
    import java.nio.channels.Selector;
    import java.nio.channels.SocketChannel;
    import java.util.Iterator;
    import java.util.Scanner;
    import java.util.Set;
    
    
    public class EchoClient {
    
        public void start() throws IOException {
    
            InetSocketAddress address =
                    new InetSocketAddress("127.0.0.1",
                            8888);
    
            // 1、获取通道（channel）
            SocketChannel socketChannel = SocketChannel.open(address);
            // 2、切换成非阻塞模式
            socketChannel.configureBlocking(false);
            //不断的自旋、等待连接完成，或者做一些其他的事情
            while (!socketChannel.finishConnect()) {
    
            }
            Print.tcfo("客户端启动成功！");
    
            //启动接受线程
            Processer processer = new Processer(socketChannel);
            new Thread(processer).start();
    
        }
    
        static class Processer implements Runnable {
            final Selector selector;
            final SocketChannel channel;
    
            Processer(SocketChannel channel) throws IOException {
                //Reactor初始化
                selector = Selector.open();
                this.channel = channel;
                channel.register(selector,
                        SelectionKey.OP_READ | SelectionKey.OP_WRITE);
            }
    
            public void run() {
                try {
                    while (!Thread.interrupted()) {
                        selector.select();
                        Set<SelectionKey> selected = selector.selectedKeys();
                        Iterator<SelectionKey> it = selected.iterator();
                        while (it.hasNext()) {
                            SelectionKey sk = it.next();
                            if (sk.isWritable()) {
                                ByteBuffer buffer = ByteBuffer.allocate(1024);
    
                                Scanner scanner = new Scanner(System.in);
                                Print.tcfo("请输入发送内容:");
                                if (scanner.hasNext()) {
                                    SocketChannel socketChannel = (SocketChannel) sk.channel();
                                    String next = scanner.next();
                                    buffer.put((Dateutil.getNow() + " >>" + next).getBytes());
                                    buffer.flip();
                                    // 操作三：通过DatagramChannel数据报通道发送数据
                                    socketChannel.write(buffer);
                                    buffer.clear();
                                }
    
                            }
                            if (sk.isReadable()) {
                                // 若选择键的IO事件是“可读”事件,读取数据
                                SocketChannel socketChannel = (SocketChannel) sk.channel();
    
                                //读取数据
                                ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                                int length = 0;
                                while ((length = socketChannel.read(byteBuffer)) > 0) {
                                    byteBuffer.flip();
                                    Logger.info("server echo:" + new String(byteBuffer.array(), 0, length));
                                    byteBuffer.clear();
                                }
    
                            }
                            //处理结束了, 这里不能关闭select key，需要重复使用
                            //selectionKey.cancel();
                        }
                        selected.clear();
                    }
                } catch (IOException ex) {
                    ex.printStackTrace();
                }
            }
        }
    
        public static void main(String[] args) throws IOException {
            new EchoClient().start();
        }
    }

#### EchoHandler ####

import util.Logger;
    
    import java.io.IOException;
    import java.nio.ByteBuffer;
    import java.nio.channels.SelectionKey;
    import java.nio.channels.Selector;
    import java.nio.channels.SocketChannel;
    
    class EchoHandler implements Runnable {
        final SocketChannel channel;
        final SelectionKey sk;
        final ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
        static final int RECIEVING = 0, SENDING = 1;
        int state = RECIEVING;
    
        EchoHandler(Selector selector, SocketChannel c) throws IOException {
            channel = c;
            c.configureBlocking(false);
            //仅仅取得选择键，后设置感兴趣的IO事件
            sk = channel.register(selector, 0);
    
            //将Handler作为选择键的附件
            sk.attach(this);
    
            //第二步,注册Read就绪事件
            sk.interestOps(SelectionKey.OP_READ);
            selector.wakeup();
        }
    
        public void run() {
    
            try {
    
                if (state == SENDING) {
                    //写入通道
                    channel.write(byteBuffer);
                    //写完后,准备开始从通道读,byteBuffer切换成写模式
                    byteBuffer.clear();
                    //写完后,注册read就绪事件
                    sk.interestOps(SelectionKey.OP_READ);
                    //写完后,进入接收的状态
                    state = RECIEVING;
                } else if (state == RECIEVING) {
                    //从通道读
                    int length = 0;
                    while ((length = channel.read(byteBuffer)) > 0) {
                        Logger.info(new String(byteBuffer.array(), 0, length));
                    }
                    //读完后，准备开始写入通道,byteBuffer切换成读模式
                    byteBuffer.flip();
                    //读完后，注册write就绪事件
                    sk.interestOps(SelectionKey.OP_WRITE);
                    //读完后,进入发送的状态
                    state = SENDING;
                }
                //处理结束了, 这里不能关闭select key，需要重复使用
                //sk.cancel();
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    
    
    }

#### EchoServerReactor ####

import java.io.IOException;
    import java.net.InetSocketAddress;
    import java.nio.channels.SelectionKey;
    import java.nio.channels.Selector;
    import java.nio.channels.ServerSocketChannel;
    import java.nio.channels.SocketChannel;
    import java.util.Iterator;
    import java.util.Set;
    
    //反应器
    class EchoServerReactor implements Runnable {
        Selector selector;
        ServerSocketChannel serverSocket;
    
        EchoServerReactor() throws IOException {
            //Reactor初始化
            selector = Selector.open();
            serverSocket = ServerSocketChannel.open();
    
            InetSocketAddress address =
                    new InetSocketAddress("127.0.0.1",
                            8888);
            serverSocket.socket().bind(address);
            //非阻塞
            serverSocket.configureBlocking(false);
    
            //分步处理,第一步,接收accept事件
            SelectionKey sk =
                    serverSocket.register(selector, SelectionKey.OP_ACCEPT);
            //attach callback object, AcceptorHandler
            sk.attach(new AcceptorHandler());
        }
    
        public void run() {
            try {
                while (!Thread.interrupted()) {
                    selector.select();
                    Set<SelectionKey> selected = selector.selectedKeys();
                    Iterator<SelectionKey> it = selected.iterator();
                    while (it.hasNext()) {
                        //Reactor负责dispatch收到的事件
                        SelectionKey sk = it.next();
                        dispatch(sk);
                    }
                    selected.clear();
                }
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    
        void dispatch(SelectionKey sk) {
            Runnable handler = (Runnable) sk.attachment();
            //调用之前attach绑定到选择键的handler处理器对象
            if (handler != null) {
                handler.run();
            }
        }
    
        // Handler:新连接处理器
        class AcceptorHandler implements Runnable {
            public void run() {
                try {
                    SocketChannel channel = serverSocket.accept();
                    if (channel != null)
                        new EchoHandler(selector, channel);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    
        /**
         * main线程, 进行处理
         * @param args
         * @throws IOException
         */
        public static void main(String[] args) throws IOException {
            new Thread(new EchoServerReactor()).start();
        }
    }

#### 单线程Reactor缺点 ####

> 单线程反应器模式中， Reactor反应器和handler处理器，都执行在同一条线程上，会有一个问题，当某个handler阻塞时，会导致其他handler也不能执行， handler可能不仅负责输入和输出处理的业务，还包括监听的Acceptor处理器，这是个非常严重的问题

### 多线程的Reactor反应器模式 ###

#### 多线程reactor反应器演进 ####

handler处理器：

> 既要使用多线程，又要尽可能高效率，则可以考虑线程池

reacotr反应器

> 考虑引入多个Selector选择器，提升选择大量通道的能力

总体来说，多线程反应器模式，大致如下:

*  将负责输入输出处理的IOHandler处理器的执行，放入独立的线程池中，这样，业务处理器线程与负责服务监听和IO事件查询的反应器线程香格里拉，避免服务器的连接监听收到阻塞
 *  将反应器线程拆分为多个子反应器线程，同时引入多个选择器，每一个SubReactor线程负责一个选择器，提高反应器管理大量连接，提升大量通道的能力

#### 多线程Reactor反应器的实践 ####

1.  引入多个选择器
2.  设计一个新的子反应器SubReactor类，一个子反应器负责查询一个选择器
3.  开启多个反应器的处理线程，一个线程负责执行一个子反应器

#### MultiThreadEchoServerReactor ####

package com.wangyg.netty.ch04.multi;
    
    import java.io.IOException;
    import java.net.InetSocketAddress;
    import java.nio.channels.SelectionKey;
    import java.nio.channels.Selector;
    import java.nio.channels.ServerSocketChannel;
    import java.nio.channels.SocketChannel;
    import java.util.Iterator;
    import java.util.Set;
    import java.util.concurrent.atomic.AtomicInteger;
    
    public class MultiThreadEchoServerReactor {
        //socketchannel
        ServerSocketChannel serverSocketChannel;
        //原子整数--next是一个原子整数
        AtomicInteger next = new AtomicInteger(0); //初始为0
        //选择器数组
        Selector[] selectors = new Selector[2]; //选择器数组，两个选择器
        //子反应器
        SubReactor[] subReactors = null;
    
        //构造器
        public MultiThreadEchoServerReactor() throws IOException {
            //初始化多个selector选择器
            selectors[0] = Selector.open(); //创建第一个选择器
            selectors[1] = Selector.open(); //创建另一个线程池
    
            serverSocketChannel = ServerSocketChannel.open(); //创建通道
    
            InetSocketAddress address =
                    new InetSocketAddress("127.0.0.1",
                            8888);
            //进行绑定通道
            serverSocketChannel.socket().bind(address);
            //设置为非阻塞
            serverSocketChannel.configureBlocking(false);
    
            //第一个选择器，负责监控新连接
    
            SelectionKey sk = serverSocketChannel.register(selectors[0], SelectionKey.OP_ACCEPT);//监听连接事件
    
            //attach附件添加
            sk.attach(new AcceptorHandler());
    
            //也是创建两个子反应器
            //创建第一个反应器
            SubReactor subReactor1 = new SubReactor(selectors[0]);
            //创建第二个反应器
            SubReactor subReactor2 = new SubReactor(selectors[1]);
            subReactors = new SubReactor[]{subReactor1, subReactor2}; //创建对应的子反应器线程数组
        }
    
        //启动服务
        private void startService() {
            new Thread(subReactors[0]).start();
            new Thread(subReactors[1]).start();
        }
    
        /**
         * handler处理器
         * 接收handler，将事件分发给handler进行处理
         */
        class AcceptorHandler implements Runnable {
            @Override
            public void run() {
                try {
                    SocketChannel channel = serverSocketChannel.accept();
                    if (channel != null) {
                        //传入  选择器和通道
                        new MultiThreadEchoHandler(selectors[next.get()], channel);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
                if (next.incrementAndGet() == selectors.length) {
                    next.set(0);
                }
            }
        }
    
        //子反应器线程 --一个子反应器对应一个selector选择器
        class SubReactor implements Runnable {
            final Selector selector;
    
            //构造函数
            public SubReactor(Selector selector) {
                this.selector = selector;
            }
    
            public void run() {
                try {
                    while (!Thread.interrupted()) {
                        selector.select();
                        Set<SelectionKey> keySet = selector.selectedKeys();
                        Iterator<SelectionKey> it = keySet.iterator();
                        while (it.hasNext()) {
                            //Reactor负责dispatch收到的事件
                            SelectionKey sk = it.next();
                            dispatch(sk);
                        }
                        keySet.clear();
                    }
                } catch (IOException ex) {
                    ex.printStackTrace();
                }
            }
    
            void dispatch(SelectionKey sk) {
                Runnable handler = (Runnable) sk.attachment();
                //调用之前attach绑定到选择键的handler处理器对象
                if (handler != null) {
                    handler.run();
                }
            }
        }
    
        public static void main(String[] args) throws IOException {
            MultiThreadEchoServerReactor server = new MultiThreadEchoServerReactor();
            server.startService();//启动service
        }
    }

#### MultiThreadEchoHandler ####

package com.wangyg.netty.ch04.multi;
    
    import util.Logger;
    
    import java.io.IOException;
    import java.nio.ByteBuffer;
    import java.nio.channels.SelectionKey;
    import java.nio.channels.Selector;
    import java.nio.channels.SocketChannel;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    /**
     * 持续性优化，因为对于多核CPU， 进行分拆为两部分，处理，这样效率更高，资源利用更充分
     */
    public class MultiThreadEchoHandler  implements  Runnable{
        final SocketChannel channel;
        final SelectionKey sk;
        final ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
        static final int RECIEVING = 0, SENDING = 1;
        int state = RECIEVING;
        //引入线程池
        static ExecutorService pool = Executors.newFixedThreadPool(4); //引入线程池
    
        MultiThreadEchoHandler(Selector selector, SocketChannel c) throws IOException {
            channel = c;
            c.configureBlocking(false);
            //仅仅取得选择键，后设置感兴趣的IO事件
            sk = channel.register(selector, 0);
            //将本Handler作为sk选择键的附件，方便事件dispatch
            sk.attach(this);
            //向sk选择键注册Read就绪事件
            sk.interestOps(SelectionKey.OP_READ);
            selector.wakeup();
        }
    
        public void run() {
            //异步任务，在独立的线程池中执行
            pool.execute(new AsyncTask());
        }
    
        //异步任务，不在Reactor线程中执行
        public synchronized void asyncRun() {
            try {
                if (state == SENDING) {
                    //写入通道
                    channel.write(byteBuffer);
                    //写完后,准备开始从通道读,byteBuffer切换成写模式
                    byteBuffer.clear();
                    //写完后,注册read就绪事件
                    sk.interestOps(SelectionKey.OP_READ);
                    //写完后,进入接收的状态
                    state = RECIEVING;
                } else if (state == RECIEVING) {
                    //从通道读
                    int length = 0;
                    while ((length = channel.read(byteBuffer)) > 0) {
                        Logger.info(new String(byteBuffer.array(), 0, length));
                    }
                    //读完后，准备开始写入通道,byteBuffer切换成读模式
                    byteBuffer.flip();
                    //读完后，注册write就绪事件
                    sk.interestOps(SelectionKey.OP_WRITE);
                    //读完后,进入发送的状态
                    state = SENDING;
                }
                //处理结束了, 这里不能关闭select key，需要重复使用
                //sk.cancel();
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    
        //异步任务的内部类
        class AsyncTask implements Runnable {
            public void run() {
                MultiThreadEchoHandler.this.asyncRun();
            }
        }
    }