Reactor模式和NIO

Bertha 。 2022-12-28 11:22 151阅读 0赞

# Reactor模式 #

不管是普通的WEB服务还是分布式服务，它们都有一个相似的结构，可以分为五个部分：

1.  接收请求：通过IO从网络中读取到请求的字节流。
2.  解码请求：将字节流转换为我们熟悉的请求对象。
3.  处理服务：通过得到的请求对象，执行一些业务逻辑。
4.  编码响应：将我们要响应的对象编码成通过网络传输的字节流。
5.  发送响应：通过IO将响应字节流发送到网络中。

通过以上五步，我们可以轻松看到，IO在网络服务中起着举足轻重的作用。

## 传统的服务设计 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center]

每个handler都有自己的一个线程，处理来自一个客户端的请求，可以想象客户端数量巨大时，创建的线程开销非常大，同时线程上下文切换的成本也很高。

package com.morris.reactor;
    
    import java.io.IOException;
    import java.net.ServerSocket;
    import java.net.Socket;
    
    // Classic Service Designs
    // 以下示例代码不考虑异常处理
    public class ClassicServerSocket implements Runnable { 
        private int port;
        private ServerSocket socket;
    
        public ClassicServerSocket(int port) throws IOException { 
            this.port = port;
            socket = new ServerSocket(port);
        }
    
        @Override
        public void run() { 
            while (!Thread.interrupted()) { 
                // 阻塞直到接收到连接请求
                Socket client = null;
                try { 
                    client = socket.accept();
                } catch (IOException e) { 
                    e.printStackTrace();
                }
                // 为该客户端的连接开一个线程
                new Thread(new Handler(client)).start();
            }
        }
    
        private class Handler implements Runnable { 
            private Socket socket;
    
            public Handler(Socket socket) { 
                this.socket = socket;
            }
    
            @Override
            public void run() { 
                // 自定义代码逻辑（对应网络服务基本的五步）
            }
        }
    }

## 高伸缩性的目标 ##

*  当负载增大时能优雅的降级服务。
 *  当资源（CPU、内存、磁盘、带宽）提升时能优化服务。
 *  低延迟。
 *  能应对徒增的请求。
 *  可以调控的服务质量。

可以用**分治法**来实现上面的目标，思路就是将一个处理过程分为很多非阻塞的多个小任务，每个任务在被允许的时候就去执行它。

以上就是一个**事件驱动**的设计，事件驱动设计使用的资源更少，不用针对每个请求启用一条线程，减少了上下文切换，减少了阻塞。但是任务分发可能会更慢，而且必须手动将事件和对应的处理动作绑定，相应的编程实现也会更难。

## Reactor模式 ##

Reacto模式就是事件驱动的设计，它按照发生的IO事件类型，将其分发到该事件对应的Handler处理，在Handler中执行的是一些非阻塞操作。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center 1]

一个线程实现的最基本的Reactor模式代码如下：

package com.morris.reactor;
    
    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.ServerSocketChannel;
    import java.nio.channels.SocketChannel;
    import java.util.Iterator;
    import java.util.Set;
    
    public class Refactor implements Runnable { 
        final Selector selector;
        final ServerSocketChannel serverSocket;
    
        public Refactor(int port) throws IOException { 
            selector = Selector.open();
            serverSocket = ServerSocketChannel.open();
            serverSocket.bind(new InetSocketAddress(port));
            serverSocket.configureBlocking(false);
            // 监听ACCEPT事件
            SelectionKey sk = serverSocket.register(selector, SelectionKey.OP_ACCEPT);
            // 绑定ACCEPT事件对应的处理单元Acceptor
            sk.attach(new Acceptor());
        }
    
        @Override
        public void run() { 
            try { 
                while (!Thread.interrupted()) { 
                    // 发生一直阻塞直到有事件
                    selector.select();
                    Set<SelectionKey> selected = selector.selectedKeys();
                    Iterator<SelectionKey> it = selected.iterator();
                    // 处理所有已发生事件
                    while (it.hasNext()) { 
                        // 将事件分发给对应的处理单元
                        dispatch(it.next());
                    }
                    selected.clear();
                }
            } catch (Exception e) { 
                e.printStackTrace();
            }
        }
    
        void dispatch(SelectionKey k) { 
            // 获得已经绑定的处理单元
            Runnable r = (Runnable) (k.attachment());
            if (r != null) { 
                // 执行处理单元
                r.run();
            }
        }
    
        class Acceptor implements Runnable { 
            @Override
            public void run() { 
                try { 
                    SocketChannel c = serverSocket.accept();
                    if (c != null) { 
                        // 用得到的socket，new一个读事件的处理单元
                        new BasicDesignHandler(selector, c);
                    }
                } catch (Exception e) { 
                    e.printStackTrace();
                }
            }
        }
    
        final class BasicDesignHandler implements Runnable { 
            private static final int MAXIN = 1024;
            private static final int MAXOUT = 1024;
            final SocketChannel socket;
            final SelectionKey sk;
            ByteBuffer input = ByteBuffer.allocate(MAXIN);
            ByteBuffer output = ByteBuffer.allocate(MAXOUT);
            static final int READING = 0, SENDING = 1;
            int state = READING;
    
            BasicDesignHandler(Selector selector, SocketChannel socket) throws IOException { 
                this.socket = socket;
                socket.configureBlocking(false);
                sk = socket.register(selector, 0);
                // 绑定本处理单元对象
                sk.attach(this);
                sk.interestOps(SelectionKey.OP_READ);
                selector.wakeup();
            }
    
            boolean inputIsComplete() { 
                /* ... */
                return true;
            }
    
            boolean outputIsComplete() { 
                /* ... */
                return true;
            }
    
            void process() { 
                /* ... */
            }
    
            @Override
            public void run() { 
                try { 
                    if (state == READING) { 
                        read();
                    } else if (state == SENDING) { 
                        send();
                    }
                } catch (Exception e) { 
                    e.printStackTrace();
                }
            }
    
            void read() throws IOException { 
                socket.read(input);
                if (inputIsComplete()) { 
                    process();
                    state = SENDING;
                    sk.interestOps(SelectionKey.OP_WRITE);
                }
            }
    
            void send() throws IOException { 
                socket.write(output);
                if (outputIsComplete()) { 
                    sk.cancel();
                }
            }
        }
    }

上面的BasicDesignHandler可以改造为如下：

final class BasicDesignHandler implements Runnable { 
        private static final int MAXIN = 1024;
        final SocketChannel socket;
        final SelectionKey sk;
        ByteBuffer input = ByteBuffer.allocate(MAXIN);
        static final int READING = 0, SENDING = 1;
        int state = READING;
    
        BasicDesignHandler(Selector selector, SocketChannel socket) throws IOException { 
            this.socket = socket;
            socket.configureBlocking(false);
            sk = socket.register(selector, 0);
            // 绑定本处理单元对象
            sk.attach(this);
            sk.interestOps(SelectionKey.OP_READ);
            selector.wakeup();
        }
    
        boolean inputIsComplete() { 
            /* ... */
            return true;
        }
    
        void process() { 
            /* ... */
        }
    
        @Override
        public void run() { 
            try { 
                socket.read(input);
                if (inputIsComplete()) { 
                    process();
                    state = SENDING;
                    sk.attach(new Sender());
                    sk.interestOps(SelectionKey.OP_WRITE);
                }
            } catch (IOException e) { 
                e.printStackTrace();
            }
        }
    
        final class Sender implements Runnable { 
    
            private static final int MAXOUT = 1024;
            ByteBuffer output = ByteBuffer.allocate(MAXOUT);
    
            boolean outputIsComplete() { 
                /* ... */
                return true;
            }
    
            @Override
            public void run() { 
                try { 
                    socket.write(output);
                    if (outputIsComplete()) { 
                        sk.cancel();
                    }
                } catch (IOException e) { 
                    e.printStackTrace();
                }
            }
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/a55e207ddef34ca4a151904a68117d5a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/182c5b93a7e941f8b1d51856b6f34120.png