10-高并发-Reactor-反应器模式-

分手后的思念是犯贱 2023-06-11 10:22 2阅读 0赞

## 一、总体来说，反应器模式类似事件驱动模式。 ##

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

## 二、反应器模式的重要组件： ##

*  Reactor反应器:负责查询IO事件，当检测到一个IO事件，将其发送给响应的Handler处理器。
 *  Handler处理器：与IO事件（或者选择键）绑定，负责IO事件的处理。完成真正的连接建立、通道读取、业务处理、写出结果。
 *  ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZG9uZ2xpYW5nemhpY2Fp_size_16_color_FFFFFF_t_70][]
 *  SelectionKey选择键的几个重要方法:
 *  void attach(Object o)---为SelectionKey添加附件。
 *  Object attachment()---获取SelectionKey的附件。

## 三、单线程反应器实例 ##

顺序：

1.  创建反应器。（打开选择器，监听端口）
2.  selector注册serverSocket---ACCEPT（得到selectionKey-事件集）
3.  selection附件：Accpt处理器。
4.  运行反应器，选择器获取事件。
5.  分发事件。
6.  获取事件集对应的附件，运行Accpt处理器。
7.  Accpt处理器：接收新连接
8.  Accpt处理器：为新连接创建IOHandler
9.  Accpt处理器：为新连接创建IOHandler  ---- (1)新的SocketChannel传输通道注册,同一个选择器中，保证是一个线程处理。
10. Accpt处理器：为新连接创建IOHandler  ---- (2)将IOhandler自身作为附件，加入选择器中。
11. Accpt处理器：为新连接创建IOHandler  ---- (3)注册事件：读&写。

class Reactor implements Runnable {
                Selector selector;
                ServerSocketChannel serverSocket;
                EchoServer Reactor() throws IOException {
                  //...．省略：打开选择器、serverSocket连接监听通道
                  //注册serverSocket的accept事件---注册后才能被KEY选中---注册事件会被选择
                  SelectionKey sk =serverSocket.register(selector, SelectionKey.OP_ACCEPT);
                  //将新连接处理器作为附件，绑定到sk选择键
                  //为事件：新建连接-绑定处理器
                  sk.attach(new AcceptorHandler());
                }
    
                public void run() {
                  //选择器轮询
                  try {
                      while (! Thread.interrupted()) {
                          //获取IO事件
                          selector.select();
                          Set selected = selector.selectedKeys();
                          Iterator it = selected.iterator();
                          while (it.hasNext()) {
                          //反应器负责dispatch收到的事件
                              SelectionKey sk=it.next();
                              //分发
                              dispatch(sk);
                          }
                          selected.clear();
                      }
                  } catch (IOException ex) { ex.printStackTrace(); }
                }
                //反应器的分发方法
                void dispatch(SelectionKey k) {
                  //获取处理器
                  Runnable handler = (Runnable) (k.attachment());
                  //调用之前绑定到选择键的handler处理器对象
                  if (handler ! = null) {
                      handler.run();
                  }
              }
              // 新连接处理器
              class AcceptorHandler implements Runnable {
                  public void run() {
                      //接受新连接
                      //需要为新连接，创建一个输入输出的handler处理器-IOHandler
                  }
              }
              //….
            }
    
    
    //一、将新的SocketChannel传输通道，注册到反应器Reactor的同一个选择器中，保证是一个线程处理。
    //二、将IOhandler自身作为附件，加入选择键中。
    
    class IOHandler implements Runnable {
                final SocketChannel channel;
                final SelectionKey sk;
                IOHandler (Selector selector, SocketChannel c) throws IOException {
                  channel = c;
                  c.configureBlocking(false);
                  //仅仅取得选择键，稍候设置感兴趣的IO事件
                  sk = channel.register(selector, 0);
                  //将Handler处理器作为选择键的附件
                  sk.attach(this);
                  //注册读写就绪事件
                  sk.interestOps(SelectionKey.OP_READ|SelectionKey.OP_WRITE);
              }
              public void run()  {
              //..．处理输入和输出
              }
            }

单线程反应器的缺点：某个Handler阻塞时，会导致其他Handler得不到执行。

四、多线程Reactor反应器模式

总体思路：

*  将负责输入输出处理的Handler的执行，放入独立线程池。
 *  反应器拆分子反应器，每个子反应器负责一个选择器，引入多个选择器。
 *  本质上-就是引入多个选择器，分发，处理。

//...．反应器
            class MultiThreadEchoServerReactor {
                ServerSocketChannelserverSocket;
                AtomicInteger next = new AtomicInteger(0);
                //选择器集合，引入多个选择器
                Selector[] selectors = new Selector[2];
                //引入多个子反应器
                SubReactor[] subReactors = null;
                MultiThreadEchoServer Reactor() throws IOException {
                  //初始化多个选择器
                  selectors[0] = Selector.open();
                  selectors[1] = Selector.open();
                  serverSocket = ServerSocketChannel.open();
                  InetSocketAddress address =
                        new InetSocketAddress(NioDemoConfig.SOCKET_SERVER_IP,
                  NioDemoConfig.SOCKET_SERVER_PORT);
                  serverSocket.socket().bind(address);
                  //非阻塞
                  serverSocket.configureBlocking(false);
                  //第一个选择器，负责监控新连接事件
                  SelectionKeysk =
                        serverSocket.register(selectors[0], SelectionKey.OP_ACCEPT);
                  //绑定Handler:attach新连接监控handler处理器到SelectionKey（选择键）
                  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();
              }
    
              //子反应器
              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()) {
                                //反应器负责dispatch收到的事件
                                SelectionKeysk = it.next();
                                 dispatch(sk);
                            }
                        keySet.clear();
                        }
                      } catch (IOException ex) {
                        ex.printStackTrace();
                        }
                  }
                  void dispatch(SelectionKeysk) {
                      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 MultiThreadEchoHandler(selectors[next.get()], channel);
                      } catch (IOException e) {
                          e.printStackTrace();
                        }
                      if (next.incrementAndGet() == selectors.length) {
                          next.set(0);
                      }
                    }
                }
                public static void main(String[] args) throws IOException {
                    MultiThreadEchoServerReactor server =
                              new MultiThreadEchoServerReactor();
                    server.startService();
                }
            }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZG9uZ2xpYW5nemhpY2Fp_size_16_color_FFFFFF_t_70]: /images/20230531/26f997224dec40d3b5971194d866ee9b.png