4.1　Reactor模式的重要性

从学习的角度来说，Reactor模式相当于高性能、高并发的一项非常重要的基础知识，只有掌握了它，才能真正理解和掌握Nginx、Redis、Netty等这些大名鼎鼎的中间件技术。

4.1.1　为什么首先学习Reactor模式

Netty本身很抽象，大量应用了设计模式。所以，学习像Netty这样的“精品中的精品”框架也是需要先从设计模式入手的，而Netty的整体架构是基于Reactor模式的。所以，学习和掌握Reactor模式，对于开始学习高并发通信（包括Netty框架）的人来说，一定是磨刀不误砍柴工。

4.1.2　Reactor模式简介

Reactor模式由Reactor线程、Handlers处理器两大角色组成，两大角色的职责分别如下：
（1）Reactor线程的职责：负责响应IO事件，并且分发到Handlers处理器。
（2）Handlers处理器的职责：非阻塞的执行业务处理逻辑。

4.1.3　多线程OIO的致命缺陷

在Java的OIO编程中，原始的网络服务器程序一般使用一个while循环不断地监听端口是否有新的连接。如果有，就调用一个处理函数来完成传输处理。

  1. while(true){
  2.     socket = accept(); //阻塞，接收连接
  3.     handle(socket) ;   //读取数据、业务处理、写入结果
  4. }

如果前一个网络连接的handle(socket)没有处理完，那么后面的新连接无法被服务端接收，于是后面的请求就会被阻塞，导致服务器的吞吐量太低。这对于服务器来说是一个严重的问题。
为了解决这个严重的连接阻塞问题，出现了一个极为经典的模式：Connection Per Thread（一个线程处理一个连接）模式。

  1. public class ConnectionPerThread implements Runnable{
  3.     @Override
  4.     public void run() {
  5.         try {
  6.             // 服务器监听socket
  7.             ServerSocket serverSocket = new ServerSocket(NioDemoConfig.SOCKET_SERVER_PORT);
  8.             while (!Thread.interrupted()){
  9.                 Socket socket = serverSocket.accept();
  10.                 // 接收到一个连接后，为socket连接，新建一个专属的处理器对象
  11.                 Handler handler = new Handler(socket);
  12.                 new Thread(handler).start();
  13.             }
  14.         } catch (IOException e) {
  15.             e.printStackTrace();
  16.         }
  17.     }
  19.     static class Handler implements Runnable{
  20.         final Socket socket;
  21.         Handler (Socket s){
  22.             socket = s;
  23.         }
  24.         @Override
  25.         public void run() {
  26.             while (true){
  27.                 try {
  28.                     byte[] input = new byte[1024];
  29.                     socket.getInputStream().read(input);
  30.                     byte[] output = null;
  31.                     socket.getOutputStream().write(output);
  32.                 } catch (IOException e) {
  33.                     e.printStackTrace();
  34.                 }
  35.             }
  36.         }
  37.     }
  38. }

在系统中，线程是比较昂贵的系统资源。如果线程的数量太多，系统将无法承受。而且，线程的反复创建、销毁、切换也需要代价。因此，在高并发的应用场景下，多线程OIO的缺陷是致命的
用Reactor模式对线程的数量进行控制，做到一个线程处理大量的连接。

4.2　单线程Reactor模式

总体来说，Reactor模式有点类似事件驱动模式。在事件驱动模式中，当有事件触发时，事件源会将事件分发到Handler（处理器），由Handler负责事件处理。Reactor模式中的反应器角色类似于事件驱动模式中的事件分发器（Dispatcher）角色
（1）Reactor：负责查询IO事件，当检测到一个IO事件时将其发送给相应的Handler处理器去处理。这里的IO事件就是NIO中选择器查询出来的通道IO事件。
（2）Handler：与IO事件（或者选择键）绑定，负责IO事件的处理，完成真正的连接建立、通道的读取、处理业务逻辑、负责将结果写到通道等。

4.2.1　什么是单线程Reactor

简单地说，Reactor和Handlers处于一个线程中执行。
image.png
SelectionKey（选择键）的几个重要的成员方法：
（1）void attach(Object o)：将对象附加到选择键。
此方法可以将任何Java POJO对象作为附件添加到SelectionKey实例。此方法非常重要，因为在单线程版本的Reactor模式实现中可以将Handler实例作为附件添加到SelectionKey实例。
（2）Object attachment()：从选择键获取附加对象。
此方法与attach(Object o)是配套使用的，其作用是取出之前通过attach(Object o)方法添加到SelectionKey实例的附加对象。这个方法同样非常重要，当IO事件发生时，选择键将被select方法查询出来，可以直接将选择键的附件对象取出。

4.2.2　单线程Reactor的参考代码

  1. package com.crazymakercircle.ReactorModel;
  2. //省略import
  3. //单线程Reactor
  4. class EchoServerReactor implements Runnable {
  5.     Selector selector;
  6.     ServerSocketChannel serverSocket;
  7.     //构造函数
  8.     EchoServerReactor() throws IOException {
  9.         //省略：打开选择器、serverSocket连接监听通道
  10.         //注册serverSocket的accept新连接接收事件
  11.         SelectionKey sk = serverSocket.register(selector, SelectionKey.OP_ACCEPT);
  12.         //将新连接处理器作为附件，绑定到sk选择键
  13.         sk.attach(new AcceptorHandler());
  14.     }
  16.     public void run() {
  17.         //选择器轮询
  18.         try {
  19.             while (!Thread.interrupted()) {
  20.                 selector.select();
  21.                 Set selected = selector.selectedKeys();
  22.                 Iterator it = selected.iterator();
  23.                 while (it.hasNext()) {
  24.                    //反应器负责dispatch收到的事件
  25.                     SelectionKey sk=it.next();
  26.                     dispatch(sk);
  27.                 }
  28.                 selected.clear();
  29.             }
  30.         } catch (IOException ex) { ex.printStackTrace(); }
  31.     }
  32.     //反应器的分发事件
  33.     void dispatch(SelectionKey k) {
  34.         Runnable handler = (Runnable) (k.attachment());
  35.         //调用之前绑定到选择键的handler对象
  36.         if (handler != null) {
  37.             handler.run();
  38.         }
  39.     }
  40. //处理器：处理新连接
  41. class AcceptorHandler implements Runnable {
  42.     public void run() {
  43.         //接受新连接
  44.         SocketChannel channel = serverSocket.accept();
  45.         //需要为新连接创建一个输入输出的handler
  46.          if (channel != null)
  47.                 new EchoHandler(selector, channel);
  48.     }
  49. }
  50. //…
  51. }

在注册serverSocket服务监听连接的接受事件之后，创建一个AcceptorHandler新连接处理器的实例作为附件，被附加（attach）到SelectionKey中。

  1. //为serverSocket注册新连接接受（accept）事件
  2. SelectionKeysk =serverSocket.register(selector, SelectionKey.OP_ACCEPT);
  3. //将新连接处理器作为附件，绑定到sk选择键
  4. sk.attach(new AcceptorHandler());
  1. 当新连接事件发生后，取出之前附加到SelectionKey中的Handler业务处理器进行socket的各种IO处理。
  1. void dispatch(SelectionKey k) {
  2.     Runnable r = (Runnable) (k.attachment());
  3.     //调用之前绑定到选择键的处理器对象
  4.     if (r != null) {
  5.        r.run();
  6.     }
  7. }

处理器AcceptorHandler的两大职责是完成新连接的接收工作、为新连接创建一个负责数据传输的Handler（称之为IOHandler）。

  1. //新连接处理器
  2. class AcceptorHandler implements Runnable {
  3.      public void run() {
  4.       //接受新连接
  5.        SocketChannel channel = serverSocket.accept();
  6.        //需要为新连接创建一个输入输出的Handler
  7.        if (channel != null)  new EchoHandler(selector, channel);
  8.      }
  9. }
  1. 顾名思义，IOHandler就是负责socket连接的数据输入、业务处理、结果输出。
  1. package com.crazymakercircle.ReactorModel;
  2. //负责数据传输的Handler
  3. class IOHandler implements Runnable {
  4.     final SocketChannel channel;
  5.     final SelectionKey sk;
  6.     IOHandler (Selector selector, SocketChannel c) {
  7.         channel = c;
  8.         c.configureBlocking(false);
  9.         //与之前的注册方式不同，先仅仅取得选择键，之后再单独设置感兴趣的IO事件
  10.         sk = channel.register(selector, 0);  //仅仅取得选择键
  11.         //将Handler处理器作为选择键的附件
  12.         sk.attach(this);
  13.         //注册读写就绪事件
  14.         sk.interestOps(SelectionKey.OP_READ|SelectionKey.OP_WRITE);
  15.     }
  16.     public void run()  {
  17.     //…处理输入和输出
  18.    }
  19. }

（1）将新的SocketChannel传输通道注册到Reactor类的同一个选择器中。这样保证了Reactor在查询IO事件时能查询到Handler注册到选择器的IO事件（数据传输事件）。
（2）Channel传输通道注册完成后，将IOHandler实例自身作为附件附加到选择键中。这样，在Reactor类分发事件（选择键）时，能执行到IOHandler的run()方法，完成数据传输处理。

4.2.3　单线程Reactor模式的EchoServer的实战案例

EchoServer的功能很简单：读取客户端的输入并回显到客户端，所以也叫回显服务器。基于Reactor模式来实现，设计三个重要的类：
（1）设计一个反应器类：EchoServerReactor类。
（2）设计两个处理器类：AcceptorHandler新连接处理器、EchoHandler回显处理器。

  1. public class EchoServerReactor implements Runnable{
  2.     Selector selector;
  3.     ServerSocketChannel serverSocketChannel;
  5.     public EchoServerReactor() throws IOException {
  6.         selector = Selector.open();
  7.         serverSocketChannel = ServerSocketChannel.open();
  8.         serverSocketChannel.bind(new InetSocketAddress("127.0.0.1", 18899));
  9.         serverSocketChannel.configureBlocking(false);
  10.         Logger.info("服务端已经开始监听：" + " 127.0.0.1 " + "18899");
  11.         SelectionKey sk = serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
  12.         // 将新连接处理器作为附件，绑定到sk选择器
  13.         sk.attach(new AcceptorHandler());
  14.     }
  16.     @Override
  17.     public void run() {
  18.         // 选择器轮询
  19.         try {
  20.             while (!Thread.interrupted()){
  21.                 selector.select();
  22.                 Set<SelectionKey> selected = selector.selectedKeys();
  23.                 Iterator<SelectionKey> it = selected.iterator();
  24.                 while (it.hasNext()){
  25.                     // 反应器负责dispatch收到的事件
  26.                     SelectionKey sk = it.next();
  27.                     dispatch(sk);
  28.                 }
  29.                 selected.clear();
  30.             }
  31.         } catch (IOException e) {
  32.             e.printStackTrace();
  33.         }
  34.     }
  36.     void dispatch(SelectionKey key){
  37.         Runnable handler = (Runnable) key.attachment();
  38.         if (handler != null){
  39.             handler.run();
  40.         }
  41.     }
  43.     // 处理器，处理新连接
  44.     class AcceptorHandler implements Runnable{
  46.         @Override
  47.         public void run() {
  48.             try {
  49.                 // 接受新连接
  50.                 SocketChannel socketChannel = serverSocketChannel.accept();
  51.                 Logger.info("接收到一个连接");
  52.                 // 需要为新连接创建一个输入输出的handler
  53.                 if (socketChannel != null){
  54.                     new EchoHandler(selector, socketChannel);
  55.                 }
  56.             } catch (IOException e) {
  57.                 e.printStackTrace();
  58.             }
  59.         }
  60.     }
  62.     public static void main(String[] args) throws IOException {
  63.         new Thread(new EchoServerReactor()).start();
  64.     }
  65. }

第二个处理器为EchoHandler回显处理器，也是一个传输处理器，主要是完成客户端的内容读取和回显

  1. public class EchoHandler implements Runnable{
  2.     final SocketChannel socketChannel;
  3.     final SelectionKey sk;
  4.     final ByteBuffer buffer = ByteBuffer.allocate(1024);
  5.     static final int RECIVING = 0, SENDING = 1;
  6.     int state = RECIVING;
  8.     public EchoHandler(Selector selector, SocketChannel c) throws IOException {
  9.         socketChannel = c;
  10.         c.configureBlocking(false);
  11.         // 与之前的注册方式不同，先仅仅取得选择键，之后再单独设置感兴趣的IO事件
  12.         sk = socketChannel.register(selector, 0);
  13.         // 将handler处理器作为选择键的附件
  14.         sk.attach(this);
  15.         // 注册读写就绪事件
  16.         sk.interestOps(SelectionKey.OP_READ);
  17.     }
  19.     @Override
  20.     public void run() {
  21.         try {
  22.             if (state == SENDING){
  23.                 //写入通道
  24.                 socketChannel.write(buffer);
  25.                 // 写完后，准备开始从通道读
  26.                 buffer.clear();
  27.                 // 写完后，注册read就绪事件
  28.                 sk.interestOps(SelectionKey.OP_READ);
  29.                 // 写完后，进入接受状态
  30.                 state = RECIVING;
  31.             }else if (state == RECIVING){
  32.                 //从通道读
  33.                 int length = 0;
  34.                 while ((length = socketChannel.read(buffer)) > 0){
  35.                     Logger.info(new String(buffer.array(), 0, length));
  36.                 }
  37.                 //读完后，准备开始写入通道,byteBuffer切换成读模式
  38.                 buffer.flip();
  39.                 //读完后，注册write就绪事件
  40.                 sk.interestOps(SelectionKey.OP_WRITE);
  41.                 //读完后,进入发送的状态
  42.                 state = SENDING;
  43.             }
  44.         } catch (IOException e) {
  45.             e.printStackTrace();
  46.         }
  47.     }
  48. }

4.2.4　单线程Reactor模式的缺点

Reactor和Handler都在同一条线程中执行。这样，带来了一个问题：当其中某个Handler阻塞时，会导致其他所有的Handler都得不到执行。在这种场景下，被阻塞的Handler不仅仅负责输入和输出处理的传输处理器，还包括负责新连接监听的AcceptorHandler处理器，可能导致服务器无响应。

4.3　多线程Reactor模式

4.3.1　多线程版本的Reactor模式演进

（1）升级Handler。既要使用多线程，又要尽可能高效率，则可以考虑使用线程池。
（2）升级Reactor。可以考虑引入多个Selector（选择器），提升选择大量通道的能力。
总体来说，多线程版本的Reactor模式大致如下：
（1）将负责数据传输处理的IOHandler处理器的执行放入独立的线程池中。这样，业务处理线程与负责新连接监听的反应器线程就能相互隔离，避免服务器的连接监听受到阻塞。
（2）如果服务器为多核的CPU，可以将反应器线程拆分为多个子反应器（SubReactor）线程；同时，引入多个选择器，并且为每一个SubReactor引入一个线程，一个线程负责一个选择器的事件轮询。这样充分释放了系统资源的能力，也大大提升了反应器管理大量连接或者监听大量传输通道的能力。

4.3.2　多线程版本Reactor的实战案例

（1）引入多个选择器。
（2）设计一个新的子反应器（SubReactor）类，子反应器负责查询一个选择器。
（3）开启多个处理线程，一个处理线程负责执行一个子反应器。
image.png

  1. package com.crazymakercircle.ReactorModel;
  2. //…
  3. //多线程版本反应器
  4. class MultiThreadEchoServerReactor {
  5.     ServerSocketChannel serverSocket;
  6.     AtomicInteger next = new AtomicInteger(0);
  7.     //选择器集合，引入多个选择器
  8.     Selector[] selectors = new Selector[2];
  9.     //引入多个子反应器
  10.     SubReactor[] subReactors = null;
  11.     MultiThreadEchoServerReactor() throws IOException {
  12.         //初始化多个选择器
  13.         selectors[0] = Selector.open(); //用于监听新连接事件
  14.         selectors[1] = Selector.open(); //用于监听传输事件
  15.         serverSocket = ServerSocketChannel.open();
  16.         InetSocketAddress address = new InetSocketAddress("127.0.0.1",18899);
  17.         serverSocket.socket().bind(address);
  18.         //非阻塞
  19.         serverSocket.configureBlocking(false);
  20.         //第一个选择器，负责监控新连接事件
  21.         SelectionKey sk = serverSocket.register(selectors[0],OP_ACCEPT);
  22.         //绑定Handler：新连接监控Handler绑定到SelectionKey（选择键）
  23.         sk.attach(new AcceptorHandler());
  24.         //第一个子反应器，负责第一个选择器的新连接事件分发（而不处理）
  25.         SubReactor subReactor1 = new SubReactor(selectors[0]);
  26.         //第二个子反应器，负责第二个选择器的传输事件的分发（而不处理）
  27.         SubReactor subReactor2 = new SubReactor(selectors[1]);
  28.         subReactors = new SubReactor[]{subReactor1, subReactor2};
  29.     }
  31.     private void startService() {
  32.         //一个子反应器对应一个线程
  33.         new Thread(subReactors[0]).start();
  34.         new Thread(subReactors[1]).start();
  35.     }
  37.     //子反应器，负责事件分发，但是不负责事件处理
  38.     class SubReactor implements Runnable {
  39.         //每个线程负责一个选择器的查询和选择
  40.         final Selector selector;
  41.         public SubReactor(Selector selector) {
  42.             this.selector = selector;
  43.         }
  44.         public void run() {
  45.             try {
  46.                 while (!Thread.interrupted()) {
  47.                     selector.select();
  48.                     Set<SelectionKey>keySet = selector.selectedKeys();
  49.                     Iterator<SelectionKey> it = keySet.iterator();
  50.                     while (it.hasNext()) {
  51.                         //dispatch所查询的事件
  52.                         SelectionKeysk = it.next();
  53.                         dispatch(sk);
  54.                     }
  55.                 keySet.clear();
  56.                 }
  57.             } catch (IOException ex) {
  58.                 ex.printStackTrace();
  59.               }
  60.         }
  61.         void dispatch(SelectionKeysk) {
  62.             Runnable handler = (Runnable) sk.attachment();
  63.             //获取之前attach绑定到选择键的handler处理器对象，执行事件处理
  64.             if (handler != null) {
  65.                 handler.run();
  66.             }
  67.         }
  68.     }
  70.     //Handler:新连接处理器
  71.     class AcceptorHandler implements Runnable {
  72.         public void run() {
  73.             try {
  74.                 SocketChannel channel = serverSocket.accept();
  75.                 //创建传输处理器，并且将传输通道注册到选择器2
  76.                 if (channel != null)
  77.                   new MultiThreadEchoHandler(selectors[1], channel);
  78.             } catch (IOException e) {
  79.                 e.printStackTrace();
  80.               }
  81.         }
  82.     }
  83.     public static void main(String[] args) throws IOException {
  84.         MultiThreadEchoServerReactor server = new MultiThreadEchoServerReactor();
  85.         server.startService();
  86.     }
  87. }

4.3.3　多线程版本Handler的实战案例

新的回显处理器为MultiThreadEchoHandler，主要的升级是引入了一个线程池（ThreadPool），使得数据传输和业务处理的代码执行在独立的线程池中，彻底地做到IO处理以及业务处理线程和反应器IO事件轮询线程的完全隔离。

  1. class MultiThreadEchoHandler implements Runnable
  2. {
  3.     final SocketChannel channel;
  4.     final SelectionKey sk;
  5.     final ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
  6.     static final int RECIEVING = 0, SENDING = 1;
  7.     int state = RECIEVING;
  8.     //引入线程池
  9.     static ExecutorService pool = Executors.newFixedThreadPool(4);
  10.     //构造器
  11.     MultiThreadEchoHandler(Selector selector, SocketChannel c) {
  12.         channel = c;
  13.         c.configureBlocking(false);
  14.         //先取得选择键，再设置感兴趣的IO事件
  15.         sk = channel.register(selector, 0);
  16.         //将本Handler作为sk选择键的附件，方便事件dispatch
  17.         sk.attach(this);
  18.         //向sk选择键设置Read就绪事件
  19.         sk.interestOps(SelectionKey.OP_READ);
  20.         selector.wakeup();
  21.     }
  22.     //此run方法在IO事件轮询线程中被调用
  23.     public void run()
  24.     {
  25.         //提交数据传输任务到线程池
  26.         //使得IO处理不在IO事件轮询线程中执行，而是在独立的线程池中执行
  27.         pool.submit(()->asyncRun());
  28.     }
  30.     //数据传输与业务处理任务，不在IO事件轮询线程中执行，而是在独立的线程池中执行
  31.     public synchronized void asyncRun()
  32.     {
  33.         try
  34.         {
  35.             if (state == SENDING)
  36.             {
  37.                 //写入通道
  38.                 channel.write(byteBuffer);
  39.                 //byteBuffer切换成写模式，写完后准备开始从通道读
  40.                 byteBuffer.clear();
  41.                 //写完后，注册read就绪事件
  42.                 sk.interestOps(SelectionKey.OP_READ);
  43.                 //进入接收的状态
  44.                 state = RECIEVING;
  45.             } else if (state == RECIEVING)
  46.             {
  47.                 //从通道读
  48.                 int length = 0;
  49.                 while ((length = channel.read(byteBuffer)) > 0)
  50.                 {
  51.                Logger.info(new String(byteBuffer.array(), 0, length));
  52.                 }
  53.                 //读完后，翻转byteBuffer的读写模式
  54.                 byteBuffer.flip();
  55.                 //注册write就绪事件
  56.                 sk.interestOps(SelectionKey.OP_WRITE);
  57.                 //进入发送的状态
  58.                 state = SENDING;
  59.             }
  60.             //处理结束了，这里不能关闭select key，需要重复使用
  61.             //sk.cancel();
  62.         } catch (IOException ex)
  63.         {
  64.             ex.printStackTrace();
  65.         }
  66.     }
  67. }

4.4　Reactor模式的优缺点

（1）Reactor模式和生产者消费者模式对比
二者的相似之处：在一定程度上，Reactor模式有点类似生产者消费者模式。在生产者消费者模式中，一个或多个生产者将事件加入一个队列中，一个或多个消费者主动从这个队列中拉取（Pull）事件来处理。
二者的不同之处：Reactor模式是基于查询的，没有专门的队列去缓冲存储IO事件，查询到IO事件之后，反应器会根据不同IO选择键（事件）将其分发给对应的Handler来处理。
（2）Reactor模式和观察者模式对比
二者的相似之处：在Reactor模式中，当查询到IO事件后，服务处理程序使用单路/多路分发（Dispatch）策略，同步分发这些IO事件。观察者模式（Observer Pattern）也被称作发布/订阅模式，它定义了一种依赖关系，让多个观察者同时监听某一个主题（Topic）。这个主题对象在状态发生变化时会通知所有观察者，它们能够执行相应的处理。
二者的不同之处：在Reactor模式中，Handler实例和IO事件（选择键）的订阅关系基本上是一个事件绑定到一个Handler，每一个IO事件（选择键）被查询后，反应器会将事件分发给所绑定的Handler，也就是一个事件只能被一个Handler处理；在观察者模式中，同一时刻、同一主题可以被订阅过的多个观察者处理。

作为高性能的IO模式，Reactor模式的优点如下：

  1. 响应快，虽然同一反应器线程本身是同步的，但是不会被单个连接的IO操作所阻塞。
  2. 编程相对简单，最大限度避免了复杂的多线程同步，也避免了多线程各个进程之间切换的开销。
  3. 可扩展，可以方便地通过增加反应器线程的个数来充分利用CPU资源。

Reactor模式的缺点如下：

  1. Reactor模式增加了一定的复杂性，因而有一定的门槛，并且不易于调试。
  2. Reactor模式依赖于操作系统底层的IO多路复用系统调用的支持，如Linux中的epoll系统调用。如果操作系统的底层不支持IO多路复用，Reactor模式不会那么高效。
  3. 在同一个Handler业务线程中，如果出现一个长时间的数据读写，就会影响这个反应器中其他通道的IO处理。