线程模型 Reactor/Proactor 的区别

关注
发布于: 2020 年 12 月 03 日
反应器(Reactor)是一种为处理服务请求并发提交到一个或者多个服务处理程序的事件设计模式，当接收请求后，服务处理程序使用解多路分配策略，然后同步地派发这些请求至相关的请求处理程序。 处理特点： 1.事件驱动event handling）
2.可以处理一个或多个输入源（one or more inputs）
3.通过Service Handler同步的将输入事件（Event）采用多路复用分发给相应的Request Handler（多个）处理
﻿
﻿
reactor简要模型png
﻿
﻿
Reactor的三种模型首先Reactor 模式中可定义三种角色：
Reactor负责监听和分配事件，将I/O事件分派给对应的Handler
Acceptor处理客户端新连接，并分派请求到处理器链中
Handlers执行非阻塞读/写 任务，可用资源池来管理
单Reactor单线程模型
﻿
﻿
Reactor 对象通过 select 监控连接事件，收到事件后通过dispatch进行转发。如果是连接建立的事件，则由Acceptor接受连接，并创建handler处理后续事件。handler会完成read->业务处理-->send的完整业务流程。
只是在代码上进行了组件的区分，但是整体操作还是单线程，不能充分利用硬件资源。handler业务处理部分没有异步。Redis使用单Reactor单进程的模型。 单Reactor多线程模型
﻿
﻿
相对于第一种单线程的模式来说，在处理业务逻辑，也就是获取到IO的读写事件之后，交由线程池来处理，handler收到响应后通过send将响应结果返回给客户端。这样可以减小主reactor的性能开销，从而更专注的做事件分发工作了，从而提升整个应用的吞吐。但是该模型存在的问题为; 1.多线程数据共享和访问比较复杂。如果子线程完成业务处理后，把结果传递给主线程reactor进行发送，就会涉及共享数据的互斥和保护机制。 2.reactor承担所有事件的监听和响应，只在主线程中运行，瞬间高并发会成为性能瓶颈。 多Reactor多线程模型
﻿
﻿
Linuxc/c++服务器开发高阶视频书学习资料+qun720209036获取﻿
﻿
关注VX公众号：Linux C后台服务器开发
﻿
﻿
第三种模型比起第二种模型，是将Reactor分成两部分： mainReactor负责监听server socket，用来处理新连接的建立，将建立的socketChannel指定注册给subReactor。 subReactor维护自己的selector, 基于mainReactor 注册的socketChannel多路分离IO读写事件，读写网 络数据，对业务处理的功能，另其扔给worker线程池来完成。实现如下:
/**
    * 多work 连接事件Acceptor,处理连接事件
    */
  class MultiWorkThreadAcceptor implements Runnable {
​
      // cpu线程数相同多work线程
      int workCount =Runtime.getRuntime().availableProcessors();
      SubReactor[] workThreadHandlers = new SubReactor[workCount];
      volatile int nextHandler = 0;
​
      public MultiWorkThreadAcceptor() {
          this.init();
      }
​
      public void init() {
          nextHandler = 0;
          for (int i = 0; i < workThreadHandlers.length; i++) {
              try {
                  workThreadHandlers[i] = new SubReactor();
              } catch (Exception e) {
              }
​
          }
      }
​
      @Override
      public void run() {
          try {
              SocketChannel c = serverSocket.accept();
              if (c != null) {// 注册读写
                  synchronized (c) {
                      // 顺序获取SubReactor，然后注册channel 
                      SubReactor work = workThreadHandlers[nextHandler];
                      work.registerChannel(c);
                      nextHandler++;
                      if (nextHandler >= workThreadHandlers.length) {
                          nextHandler = 0;
                      }
                  }
              }
          } catch (Exception e) {
          }
      }
  }
  /**
    * 多work线程处理读写业务逻辑
    */
  class SubReactor implements Runnable {
      final Selector mySelector;
​
      //多线程处理业务逻辑
      int workCount =Runtime.getRuntime().availableProcessors();
      ExecutorService executorService = Executors.newFixedThreadPool(workCount);
​
​
      public SubReactor() throws Exception {
          // 每个SubReactor 一个selector 
          this.mySelector = SelectorProvider.provider().openSelector();
      }
​
      /**
        * 注册chanel
        *
        * @param sc
        * @throws Exception
        */
      public void registerChannel(SocketChannel sc) throws Exception {
          sc.register(mySelector, SelectionKey.OP_READ | SelectionKey.OP_CONNECT);
      }
​
      @Override
      public void run() {
          while (true) {
              try {
              //每个SubReactor 自己做事件分派处理读写事件
                  selector.select();
                  Set<SelectionKey> keys = selector.selectedKeys();
                  Iterator<SelectionKey> iterator = keys.iterator();
                  while (iterator.hasNext()) {
                      SelectionKey key = iterator.next();
                      iterator.remove();
                      if (key.isReadable()) {
                          read();
                      } else if (key.isWritable()) {
                          write();
                      }
                  }
​
              } catch (Exception e) {
​
              }
          }
      }
​
      private void read() {
          //任务异步处理
          executorService.submit(() -> process());
      }
​
      private void write() {
          //任务异步处理
          executorService.submit(() -> process());
      }
​
      /**
        * task 业务处理
        */
      public void process() {
          //do IO ,task,queue something
      }
  }
​
﻿
mainReactor 主要是用来处理网络IO 连接建立操作，通常一个线程就可以处理，而subReactor主要做和建立起来的socket做数据交互和事件业务处理操作，它的个数上一般是和CPU个数等同，每个subReactor一个线程来处理。如netty就是采用这种实现。
Reactor/Proactor模型处理read方式上的区别: Reactor： 某个事件处理者宣称它对某个socket上的读事件很感兴趣; 事件分离者等着这个事件的发生; 当事件发生了，事件分离器被唤醒，这负责通知先前那个事件处理者; 事件处理者收到消息，于是去那个socket上读数据了。 如果需要，它再次宣称对这个socket上的读事件感兴 趣，一直重复上面的步骤; Proactor： 事件处理者直接投递发一个写操作(当然，操作系统必须支持这个异步操作)。 这个时候，事件处理者根本不关心读事件，它只管发这么个请求，它魂牵梦萦的是这个写操作的完成事件。这个处理者很拽，发个命令就不管具体的事情了，只等着别人（系统）帮他搞定的时候给他回个话。 事件分离者等着这个读事件的完成(比较下与Reactor的不同); 当事件分离者默默等待完成事情到来的同时，操作系统已经在一边开始干活了，它从目标读取数据，放入用户提供的缓存区中，最后通知事件分离者，这个事情我搞完了; 事件分离者通知之前的事件处理者: 你吩咐的事情搞定了; 事件处理者这时会发现想要读的数据已经乖乖地放在他提供的缓存区中，想怎么处理都行了。如果有需要，事件处理者还像之前一样发起另外一个写操作，和上面的几个步骤一样。
﻿
﻿
procator.png
Procator Initiator负责创建Procator 和 Handler，并将Procator和Handler都通过Asynchronous operation processor注册到内核。 Asynchronous operation processor负责处理注册请求，并完成IO操作。完成IO操作后会通知 procator。 procator根据不同的事件类型回调不同的handler进行业务处理 handler完成业务处理，handler也可以注册新的handler到内核进程。
﻿
异步IO需要系统层面的支持，目前 Windows 下通过 IOCP 实现了真正的异步IO,而在 Linux 系统下的 AIO 并不完善。
﻿