三、Reactor反应器模式

       在OIO模型中，最初的程序设计是监听端口是否有连接，如果有就调用一个函数去处理。这个模式的最大的问题就是，前面的函数没有处理完，会导致后来的连接无法被接受，于是后面的请求都会被阻塞住，服务器的吞吐量被极大的限制。后来采用了Connection Per Thread（一个线程处理一个连接）模式。解决了前面的新连接被严重阻塞的问题。但是对于大量的连接，需要耗费大量的线程资源。并且线程的反复创建、销毁、线程的切换也需要代价。所以为了解决这个缺陷，一个有效的解决办法是：使用Reactor反应器模式。使用反应器模式对线程的数量进行控制，做到一个线程处理大量的连接。

       反应器设计模式(Reactorpattern)是一种为处理服务请求并发提交到一个或者多个服务处理程序的事件设计模式。当请求抵达后，服务处理程序使用解多路分配策略，然后同步地派发这些请求至相关的请求处理程序。

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

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

一、单线程Reactor反应器模式

基于Java NIO，如何实现简单的单线程版本的反应器模式呢，需要用到SelectionKey选择器的几个重要的成员方法：

1. void attach（Object obj）此方法可以将任何的Java POJO对象作为附件添加到SelectionKey实例，相当于附件属性的setter方法。
2. Object attachment（）此方法的作用是取出之前通过attach方法添加的SelectionKey选择键实例的附件，相当于附件属性的getter方法。

看一个简单的示例：

客户端：

public class EchoClient {

    public void start() throws IOException {

        InetSocketAddress address = new InetSocketAddress("127.0.0.1", 8878);

        // 1、获取通道（channel）
        SocketChannel socketChannel = SocketChannel.open(address);
        // 2、切换成非阻塞模式
        socketChannel.configureBlocking(false);
        //不断的自旋、等待连接完成，或者做一些其他的事情
        while (!socketChannel.finishConnect()) {

        }
        System.out.println("客户端启动成功！");

        //启动接受线程
        Processor processor = new Processor(socketChannel);
        new Thread(processor).start();

    }

    static class Processor implements Runnable {
        final Selector selector;
        final SocketChannel channel;

        Processor(SocketChannel channel) throws IOException {
            //Reactor初始化
            selector = Selector.open();
            this.channel = channel;
            channel.register(selector, 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();
                        dispatch(sk);
                    }
                    selected.clear();
                }
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    }


    private static void dispatch(SelectionKey selectionKey) {
        try {
            if (selectionKey.isWritable()) {
                ByteBuffer buffer = ByteBuffer.allocate(1024);
                Scanner scanner = new Scanner(System.in);
                System.out.println("请输入发送内容:");
                if (scanner.hasNext()) {
                    SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
                    String next = scanner.next();
                    buffer.put(next.getBytes());
                    buffer.flip();
                    socketChannel.write(buffer);
                    buffer.clear();
                }
                selectionKey.interestOps(SelectionKey.OP_READ);

            }
            if (selectionKey.isReadable()) {
                // 读取数据
                SocketChannel socketChannel = (SocketChannel) selectionKey.channel();

                //读取数据
                ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                int length;
                while ((length = socketChannel.read(byteBuffer)) > 0) {
                    byteBuffer.flip();
                    System.out.println("服务端返回" + new String(byteBuffer.array(), 0, length));
                    byteBuffer.clear();
                }
                selectionKey.interestOps(SelectionKey.OP_WRITE);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    public static void main(String[] args) throws IOException {
        new EchoClient().start();
    }
}

处理器：

public class EchoHandler implements Runnable {
    private final SocketChannel channel;
    private final SelectionKey sk;
    private final ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
    private static final int RECEIVING = 0, SENDING = 1;
    private int state = RECEIVING;

    public 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 = RECEIVING;
            } else if (state == RECEIVING) {
                //从通道读
                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();
        }
    }
}

反应器：

public class EchoServerReactor implements Runnable {
    private Selector selector;
    private ServerSocketChannel serverSocket;

    public EchoServerReactor() throws IOException {
        //Reactor初始化
        selector = Selector.open();
        serverSocket = ServerSocketChannel.open();

        InetSocketAddress address = new InetSocketAddress("127.0.0.1", 8878);
        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();
            }
        }
    }


    public static void main(String[] args) throws IOException {
        new Thread(new EchoServerReactor()).start();
    }
}

       单线程的反应器模式有这样一个缺陷，当其中某个Handler阻塞时，会导致其他所有的Handler都得不到执行。如果负责监听的AcceptorHandler处理器阻塞会导致整个服务都不能接受新的连接。另外服务器一般都是多核的，单线程的反应器模式未能充分利用这个资源。

二、多线程Reactor反应器模式

优化考虑：

1. 升级Handler处理器，考虑使用线程池。将负责输入输出处理的IOHandler处理器的执行，放入独立的线程池中。这样业务处理线程与负责监听和IO事件查询的反应器线程相隔离，避免服务器监听线程收到阻塞。
2. 升级Selector反应器，考虑引入多个Selector选择器，每个SubReactor负责一个选择器。

                        

下面是一个简单的Demo，目的是客户端发送一个消息到服务端，服务端直接将该信息再反馈给客户端，客户端可以采用上面的代码。

反应器：

public class MultiThreadEchoServerReactor {
    private ServerSocketChannel serverSocket;
    private AtomicInteger next = new AtomicInteger(0);
    //选择器集合，引入多个选择器
    private Selector[] selectors = new Selector[2];
    //引入多个子反应器
    private SubReactor[] subReactors;

    public MultiThreadEchoServerReactor() throws IOException {
        //初始化多个选择器
        selectors[0] = Selector.open();
        selectors[1] = Selector.open();
        serverSocket = ServerSocketChannel.open();
        InetSocketAddress address = new InetSocketAddress("127.0.0.1", 8878);
        serverSocket.socket().bind(address);
        //设置非阻塞
        serverSocket.configureBlocking(false);
        //第一个选择器负责监控新连接事件
        SelectionKey selectionKey = serverSocket.register(selectors[0], SelectionKey.OP_ACCEPT);

        //绑定Handler
        selectionKey.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();
    }


    //子反应器
    class SubReactor implements Runnable {
        //每个线程负责一个选择器的查询和选择
        private final Selector selector;

        SubReactor(Selector selector) {
            this.selector = selector;
        }

        @Override
        public void run() {
            try {
                while (!Thread.interrupted()) {
                    selector.select();
                    Set<SelectionKey> keySet = selector.selectedKeys();
                    Iterator<SelectionKey> iterator = keySet.iterator();
                    while (iterator.hasNext()) {
                        //反应器负责dispatch收到的事件
                        SelectionKey key = iterator.next();
                        dispatch(key);
                    }
                    keySet.clear();
                }
            } catch (Exception e) {
                e.printStackTrace();
            }

        }

        void dispatch(SelectionKey selectionKey) {
            Runnable handler = (Runnable) selectionKey.attachment();
            //调用之前attach绑定到选择键的Handler处理器对象
            if (handler != null) {
                handler.run();
            }
        }
    }

    //新连接处理器
    class AcceptorHandler implements Runnable {

        @Override
        public void run() {
            try {
                SocketChannel channel = serverSocket.accept();
                if (channel != null) {
                    new MultiThreadEchoHandler(selectors[next.get()], channel);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
            if (next.incrementAndGet() == selectors.length) {
                next.set(0);
            }
        }
    }

    public static void main(String[] args) throws IOException {
        MultiThreadEchoServerReactor serverReactor = new MultiThreadEchoServerReactor();
        serverReactor.startService();
    }
}

 处理器：

public class MultiThreadEchoHandler implements Runnable {
    private final SocketChannel socketChannel;
    private final SelectionKey selectionKey;
    private final ByteBuffer buffer = ByteBuffer.allocate(1024);
    private static final int RECEIVING = 0, SENDING = 1;
    private int state = RECEIVING;

    //引入线程池
    private static ExecutorService pool = Executors.newFixedThreadPool(4);

    public MultiThreadEchoHandler(Selector selector, SocketChannel channel) throws IOException {
        this.socketChannel = channel;
        channel.configureBlocking(false);
        selector.wakeup();
        //取得选择键，在设置感兴趣的IO事件
        selectionKey = channel.register(selector, SelectionKey.OP_READ，this);
    }

    @Override
    public void run() {
        //异步任务，在独立的线程池中执行
        pool.execute(new AsyncTask());
    }

    //业务处理，不在反应器线程中执行
    public synchronized void asyncRun() {
        try {
            if (state == SENDING) {
                //写入通道
                socketChannel.write(buffer);
                //写完后，准备从通道开始读，buffer切换成写入模式
                buffer.clear();
                //写完后注册read就绪事件
                selectionKey.interestOps(SelectionKey.OP_READ);
                //写完后，进入接收到的状态
                state = RECEIVING;
            } else if (state == RECEIVING) {
                //从通道读
                int length;
                while ((length = socketChannel.read(buffer)) > 0) {
                    Logger.info(new String(buffer.array(), 0, length));
                }
                //读完之后，开始准备写入通道，buffer切换成读取模式
                buffer.flip();
                //读完后，注册writer就绪事件
                selectionKey.interestOps(SelectionKey.OP_WRITE);
                //读完后进入发送的状态
                state = SENDING;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    //异步任务的内部类
    class AsyncTask implements Runnable {

        @Override
        public void run() {
            MultiThreadEchoHandler.this.asyncRun();
        }
    }
}