目录

总结 送学习资料（包含视频、面试、面试题、技术学习路线图谱、文档等）

1.框架的构建：多个技术点的平衡再选择；

2. 框架的实现：合理的拆分以及理解本质后的抽象；

3. 网络模块实现：reactor 与 协程融合形成同步非阻塞；

4. 连接池实现：异步连接 与 同步连接池的实现方式；

一、Reactor 是什么

详细教程资料+课件 关注+后台私信；资料；两个字可以免费视频领取+文档+各大厂面试题 资料内容包括：C/C++，Linux，golang,Nginx，ZeroMQ，MySQL，Redis，fastdfs，MongoDB，ZK，流媒体，CDN，P2P，K8S，Docker，TCP/IP，协程，DPDK，嵌入式 等。

关于re。actor 是什么，我们先从wiki上看下:

The reactor design pattern is an event handling pattern for handling service requests delivered concurrently to a service handler by one or more inputs. The service handler then demultiplexes the incoming requests and dispatches them synchronously to the associated request handlers.

从上述文字中我们可以看出以下关键点 ：

事件驱动（event handling）

可以处理一个或多个输入源（one or more inputs）

通过Service Handler同步的将输入事件（Event）采用多路复用分发给相应的Request Handler（多个）处理

自POSA2 中的关于Reactor Pattern 介绍中，我们了解了Reactor 的处理方式:

同步的等待多个事件源到达（采用select()实现）

将事件多路分解以及分配相应的事件服务进行处理，这个分派采用server集中处理（dispatch）

分解的事件以及对应的事件服务应用从分派服务中分离出去（handler）

关于Reactor Pattern 的OMT 类图设计:

详细教程资料+课件 关注+后台私信；资料；两个字可以免费视频领取+文档+各大厂面试题 资料内容包括：C/C++，Linux，golang,Nginx，ZeroMQ，MySQL，Redis，fastdfs，MongoDB，ZK，流媒体，CDN，P2P，K8S，Docker，TCP/IP，协程，DPDK，嵌入式 等。

二、为何要用Reactor

常见的网络服务中，如果每一个客户端都维持一个与登陆服务器的连接。那么服务器将维护多个和客户端的连接以出来和客户端的contnect 、read、write ，特别是对于长链接的服务，有多少个c端，就需要在s端维护同等的IO连接。这对服务器来说是一个很大的开销。

1. 举个例子吧

??假设我们需要建立一个提供分布式日志服务的事件驱动服务器。客户们使用客户端会向该服务器发送请求记录自己的状态信息，信息包括错误通知、debug信息、表现诊断等。该日志服务器对于收到的信息进行分类并分发操作，包括：显示屏显示、打印机打印、数据库存储等，如下图所示。

详细教程资料+课件 关注+后台私信；资料；两个字可以免费视频领取+文档+各大厂面试题 资料内容包括：C/C++，Linux，golang,Nginx，ZeroMQ，MySQL，Redis，fastdfs，MongoDB，ZK，流媒体，CDN，P2P，K8S，Docker，TCP/IP，协程，DPDK，嵌入式 等。

客户端和服务器之间通信协议通常选用TCP等面向连接的协议（为了保证可靠性），通过IP和端口的四元组来认定客户端和服务器。日志服务器被多个客户端同时使用，为此该服务器需要保证多用户连接请求和日志记录的并发性。为了保证这个基本要求，不难想到使用多线程去实现该服务器，每一个线程专门针对一个连接。

??然而，该设计存在以下的问题：

（1） 效率问题。

??多线程导致的上下文切换、同步、数据移动等可能带来效率的下降。

（2） 编程的简单性。

??多线程需要考虑复杂的并发设计，包括线程安全等。

（3）可移植性

??多线程在不同OS下的不同（或者没有）影响了可移植性。

??由于以上问题，多线程设计往往既不是最高效也不是最易实现的方案。由此我们需要别的方案来实现可以处理并发请求的服务器（推销自己的reactor）。

三、Reactor

在应用Java NIO构建Reactor Pattern中，大神 Doug Lea（让人无限景仰的java 大神）在“Scalable IO in Java”中给了很好的阐述。我们采用大神介绍的3种Reactor 来分别介绍。

首先我们基于Reactor Pattern 处理模式中，定义以下三种角色:

Reactor 将I/O事件分派给对应的Handler

Acceptor 处理客户端新连接，并分派请求到处理器链中

Handlers 执行非阻塞读/写 任务

1、单Reactor单线程模型

详细教程资料+课件 关注+后台私信；资料；两个字可以免费视频领取+文档+各大厂面试题 资料内容包括：C/C++，Linux，golang,Nginx，ZeroMQ，MySQL，Redis，fastdfs，MongoDB，ZK，流媒体，CDN，P2P，K8S，Docker，TCP/IP，协程，DPDK，嵌入式 等。

我们看代码的实现方式:

 /**
    * 等待事件到来，分发事件处理
    */
  class Reactor implements Runnable {

      private Reactor() throws Exception {
?
          SelectionKey sk =
                  serverSocket.register(selector,
                          SelectionKey.OP_ACCEPT);
          // attach Acceptor 处理新连接
          sk.attach(new Acceptor());
      }
?
      public void run() {
          try {
              while (!Thread.interrupted()) {
                  selector.select();
                  Set selected = selector.selectedKeys();
                  Iterator it = selected.iterator();
                  while (it.hasNext()) {
                      it.remove();
                      //分发事件处理
                      dispatch((SelectionKey) (it.next()));
                  }
              }
          } catch (IOException ex) {
              //do something
          }
      }
?
      void dispatch(SelectionKey k) {
          // 若是连接事件获取是acceptor
          // 若是IO读写事件获取是handler
          Runnable runnable = (Runnable) (k.attachment());
          if (runnable != null) {
              runnable.run();
          }
      }
?
  }
  
  /**
    * 连接事件就绪,处理连接事件
    */
  class Acceptor implements Runnable {
      @Override
      public void run() {
          try {
              SocketChannel c = serverSocket.accept();
              if (c != null) {// 注册读写
                  new Handler(c, selector);
              }
          } catch (Exception e) {
?
          }
      }
  }

这是最基本的单Reactor单线程模型。其中Reactor线程，负责多路分离套接字，有新连接到来触发connect 事件之后，交由Acceptor进行处理，有IO读写事件之后交给hanlder 处理。

Acceptor主要任务就是构建handler ，在获取到和client相关的SocketChannel之后 ，绑定到相应的hanlder上，对应的SocketChannel有读写事件之后，基于racotor 分发,hanlder就可以处理了（所有的IO事件都绑定到selector上，有Reactor分发）。

2、单Reactor多线程模型

详细教程资料+课件 关注+后台私信；资料；两个字可以免费视频领取+文档+各大厂面试题 资料内容包括：C/C++，Linux，golang,Nginx，ZeroMQ，MySQL，Redis，fastdfs，MongoDB，ZK，流媒体，CDN，P2P，K8S，Docker，TCP/IP，协程，DPDK，嵌入式 等。

相对于第一种单线程的模式来说，在处理业务逻辑，也就是获取到IO的读写事件之后，交由线程池来处理，这样可以减小主reactor的性能开销，从而更专注的做事件分发工作了，从而提升整个应用的吞吐。

我们看下实现方式：

/**
    * 多线程处理读写业务逻辑
    */
  class MultiThreadHandler implements Runnable {
      public static final int READING = 0, WRITING = 1;
      int state;
      final SocketChannel socket;
      final SelectionKey sk;
?
      //多线程处理业务逻辑
      ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
?
?
      public MultiThreadHandler(SocketChannel socket, Selector sl) throws Exception {
          this.state = READING;
          this.socket = socket;
          sk = socket.register(selector, SelectionKey.OP_READ);
          sk.attach(this);
          socket.configureBlocking(false);
      }
?
      @Override
      public void run() {
          if (state == READING) {
              read();
          } else if (state == WRITING) {
              write();
          }
      }
?
      private void read() {
          //任务异步处理
          executorService.submit(() -> process());
?
          //下一步处理写事件
          sk.interestOps(SelectionKey.OP_WRITE);
          this.state = WRITING;
      }
?
      private void write() {
          //任务异步处理
          executorService.submit(() -> process());
?
          //下一步处理读事件
          sk.interestOps(SelectionKey.OP_READ);
          this.state = READING;
      }
?
      /**
        * task 业务处理
        */
      public void process() {
          //do IO ,task,queue something
      }
  }
 

3、多Reactor多线程模型

详细教程资料+课件 关注+后台私信；资料；两个字可以免费视频领取+文档+各大厂面试题 资料内容包括：C/C++，Linux，golang,Nginx，ZeroMQ，MySQL，Redis，fastdfs，MongoDB，ZK，流媒体，CDN，P2P，K8S，Docker，TCP/IP，协程，DPDK，嵌入式 等。

第三种模型比起第二种模型，是将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
      }
  }

四、事件处理模式

在 Douglas Schmidt 的大作《POSA2》中有关于事件处理模式的介绍，其中有四种事件处理模式:

Reactor　　

Proactor　　

Asynchronous Completion Token　　

Acceptor-Connector　　

1.Proactor

本文介绍的Reactor就是其中一种，而Proactor的整体结构和reacotor的处理方式大同小异，不同的是Proactor采用的是异步非阻塞IO的方式实现，对数据的读写由异步处理，无需用户线程来处理，服务程序更专注于业务事件的处理，而非IO阻塞。

2.Asynchronous Completion Token

简单来说，ACT就是应对应用程序异步调用服务操作，并处理相应的服务完成事件。从token这个字面意思，我们大概就能了解到，它是一种状态的保持和传递。

比如，通常应用程序会有调用第三方服务的需求，一般是业务线程请求都到，需要第三方资源的时候，去同步的发起第三方请求，而为了提升应用性能，需要异步的方式发起请求，但异步请求的话，等数据到达之后，此时的我方应用程序的语境以及上下文信息已经发生了变化，你没办法去处理。

ACT 解决的就是这个问题，采用了一个token的方式记录异步发送前的信息，发送给接受方，接受方回复的时候再带上这个token，此时就能恢复业务的调用场景。

总结；

上图中我们可以看到在client processing 这个阶段，客户端是可以继续处理其他业务逻辑的，不是阻塞状态。service 返回期间会带上token信息。

详细教程资料+课件 关注+后台私信；资料；两个字可以免费视频领取+文档+各大厂面试题 资料内容包括：C/C++，Linux，golang,Nginx，ZeroMQ，MySQL，Redis，fastdfs，MongoDB，ZK，流媒体，CDN，P2P，K8S，Docker，TCP/IP，协程，DPDK，嵌入式 等。