文档章节

【NIO系列】——之Reactor模型

wier
 wier
发布于 07/11 10:50
字数 3589
阅读 887
收藏 33
点赞 2
评论 5

 

在开篇之前,我们对JavaNIO 的使用方式不做过多介绍,这种API的介绍方式网上太多了,没必要详细介绍,我们假设NIO的使用方式,你能够熟练运用。这是NIO系列第三篇:

【NIO系列】——之TCP探秘

【NIO系列】——之IO模型

通过之前的Unix的IO模型介绍,想必也了解到了5种IO模型。java的NIO是属于同步非阻塞IO,关于IO多路复用,java没有相应的IO模型,但有相应的编程模式,Reactor 就是基于NIO中实现多路复用的一种模式。本文将从以下几点阐述Reactor模式:

  1. reactor 是什么

  2. 为何要用,能解决什么问题

  3. 如何用,更好的方式

  4. 其他事件处理模式

一、Reactor 是什么

关于reactor 是什么,我们先从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.

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

  1. 事件驱动(event handling)

  2. 可以处理一个或多个输入源(one or more inputs)

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

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

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

  2. 将事件多路分解以及分配相应的事件服务进行处理,这个分派采用server集中处理(dispatch)

  3. 分解的事件以及对应的事件服务应用从分派服务中分离出去(handler)

 

关于Reactor Pattern 的OMT 类图设计:

 

二、为何要用Reactor

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

1、BIO

比如我们采用BIO的方式来维护和客户端的连接:

// 主线程维护连接
  public void run() {
      try {
          while (true) {
              Socket socket = serverSocket.accept();
              //提交线程池处理
              executorService.submit(new Handler(socket));
          }
      } catch (Exception e) {
          e.printStackTrace();
      }
  }
​
  // 处理读写服务
  class Handler implements Runnable {
      public void run() {
          try {
              //获取Socket的输入流,接收数据
              BufferedReader buf = new BufferedReader(new InputStreamReader(socket.getInputStream()));
              String readData = buf.readLine();
              while (readData != null) {
                  readData = buf.readLine();
                  System.out.println(readData);
              }
          } catch (Exception e) {
              e.printStackTrace();
          }
      }
  }

 

很明显,为了避免资源耗尽,我们采用线程池的方式来处理读写服务。但是这么做依然有很明显的弊端:

  1. 同步阻塞IO,读写阻塞,线程等待时间过长

  2. 在制定线程策略的时候,只能根据CPU的数目来限定可用线程资源,不能根据连接并发数目来制定,也就是连接有限制。否则很难保证对客户端请求的高效和公平。

  3. 多线程之间的上下文切换,造成线程使用效率并不高,并且不易扩展

  4. 状态数据以及其他需要保持一致的数据,需要采用并发同步控制

2、NIO

那么可以有其他方式来更好的处理么,我们可以采用NIO来处理,NIO中支持的基本机制:

  1. 非阻塞的IO读写

  2. 基于IO事件进行分发任务,同时支持对多个fd的监听

我们看下NIO 中实现相关方式:

public NIOServer(int port) throws Exception {
      selector = Selector.open();
      serverSocket = ServerSocketChannel.open();
      serverSocket.socket().bind(new InetSocketAddress(port));
      serverSocket.configureBlocking(false);
      serverSocket.register(selector, SelectionKey.OP_ACCEPT);
  }
​
  @Override
  public void run() {
      while (!Thread.interrupted()) {
          try {
              //阻塞等待事件
              selector.select();
              // 事件列表
              Set selected = selector.selectedKeys();
              Iterator it = selected.iterator();
              while (it.hasNext()) {
                  it.remove();
                  //分发事件
                  dispatch((SelectionKey) (it.next()));
              }
          } catch (Exception e) {
​
          }
      }
  }
​
  private void dispatch(SelectionKey key) throws Exception {
      if (key.isAcceptable()) {
          register(key);//新链接建立,注册
      } else if (key.isReadable()) {
          read(key);//读事件处理
      } else if (key.isWritable()) {
          wirete(key);//写事件处理
      }
  }
​
  private void register(SelectionKey key) throws Exception {
      ServerSocketChannel server = (ServerSocketChannel) key
              .channel();
      // 获得和客户端连接的通道
      SocketChannel channel = server.accept();
      channel.configureBlocking(false);
      //客户端通道注册到selector 上
      channel.register(this.selector, SelectionKey.OP_READ);
  }

 

我们可以看到上述的NIO例子已经差不多拥有reactor的影子了

  1. 基于事件驱动-> selector(支持对多个socketChannel的监听)

  2. 统一的事件分派中心-> dispatch

  3. 事件处理服务-> read & write

 

事实上NIO已经解决了上述BIO暴露的1&2问题了,服务器的并发客户端有了量的提升,不再受限于一个客户端一个线程来处理,而是一个线程可以维护多个客户端(selector 支持对多个socketChannel 监听)。

但这依然不是一个完善的Reactor Pattern ,首先Reactor 是一种设计模式,好的模式应该是支持更好的扩展性,显然以上的并不支持,另外好的Reactor Pattern 必须有以下特点:

  1. 更少的资源利用,通常不需要一个客户端一个线程

  2. 更少的开销,更少的上下文切换以及locking

  3. 能够跟踪服务器状态

  4. 能够管理handler 对event的绑定

那么好的Reactor Pattern应该是怎样的?

三、Reactor

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

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

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

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

  • Handlers 执行非阻塞读/写 任务

 

1、单Reactor单线程模型

我们看代码的实现方式:

  /**
    * 等待事件到来,分发事件处理
    */
  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多线程模型

相对于第一种单线程的模式来说,在处理业务逻辑,也就是获取到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多线程模型

 

第三种模型比起第二种模型,是将Reactor分成两部分,

  1. mainReactor负责监听server socket,用来处理新连接的建立,将建立的socketChannel指定注册给subReactor。

  2. 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一个县城来处理。

此种模型中,每个模块的工作更加专一,耦合度更低,性能和稳定性也大量的提升,支持的可并发客户端数量可达到上百万级别。

关于此种模型的应用,目前有很多优秀的矿建已经在应用了,比如mina 和netty 等。上述中去掉线程池的第三种形式的变种,也 是Netty NIO的默认模式。下一节我们将着重讲解netty的架构模式。

 

四、事件处理模式

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

  1. Reactor  

  2. Proactor  

  3. Asynchronous Completion Token  

  4. Acceptor-Connector  

1. Proactor

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

2. Asynchronous Completion Token

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

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

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

 

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

3. Acceptor-Connector

Acceptor-Connector是于Reactor的结合,也可以看成是一种变种,它看起来很像上面介绍的Reactor第三种实现方式,但又有本质的不同。

Acceptor-Connector模式是将网络中对等服务的连接和初始化分开处理,使系统中的连接建立及服务一旦服务初始化后就分开解除耦合。连接器主动地建立到远地接受器组件的连接,并初始化服务处理器来处理在连接上交换的数据。同样地,接受器被动地等待来自远地连接器的连接请求,在这样的请求到达时建立连接,并初始化服务处理器来处理在连接上交换的数据。随后已初始化的服务处理器执行应用特有的处理,并通过连接器和接受器组件建立的连接来进行通信。

这么处理的好处是:

  1. 一般而言,用于连接建立和服务初始化的策略变动的频度要远小于应用服务实现和通信协议。

  2. 容易增加新类型的服务、新的服务实现和新的通信协议,而又不影响现有的连接建立和服务初始化软件。比如采用IPX/SPX通信协议或者TCP协议。

  3. 连接角色和通信角色的去耦合,连接角色只管发起连接 vs. 接受连接。通信角色只管数据交互。

  4. 将程序员与低级网络编程API(像socket或TLI)类型安全性的缺乏屏蔽开来。业务开发关系底层通信

 

引用:

http://www.kuqin.com/ace-2002-12/Part-One/Chapter-9.htm

http://www.dre.vanderbilt.edu/%7Eschmidt/PDF/reactor-siemens.pdf

http://gee.cs.oswego.edu/dl/cpjslides/nio.pdf

 

更多架构知识,欢迎关注我的公众号,大码候(cool_wier)

© 著作权归作者所有

共有 人打赏支持
wier
粉丝 716
博文 49
码字总数 131483
作品 0
东城
技术主管
加载中

评论(5)

蓝天bluesky
蓝天bluesky
赞,讲的清楚明白
wier
wier

引用来自“Spring狂人”的评论

Reactor  有什么应用场景?
tomcat7 和netty都在用这种模式,应用场景就是可支持IO高并发
Spring狂人
Spring狂人
Reactor  有什么应用场景?
wier
wier

引用来自“士别三日”的评论

抄袭我的文章也要说一句吧
1、你发下链接,如果有抄袭,我发文致歉。
2、我承认有两篇引用没有做说明,但是英文文章,这篇写作时间长,中间丢几次就没找到了。
士别三日
士别三日
抄袭我的文章也要说一句吧
Netty与Reactor模式详解

在学习Reactor模式之前,我们需要对“I/O的四种模型”以及“什么是I/O多路复用”进行简单的介绍,因为Reactor是一个使用了同步非阻塞的I/O多路复用机制的模式。 I/O的四种模型 I/0 操作 主要...

hutaishi
昨天
0
0
服务器模型——从单线程阻塞到多线程非阻塞(下)

前言的前言 服务器模型涉及到线程模式和IO模式,搞清楚这些就能针对各种场景有的放矢。该系列分成三部分: 单线程/多线程阻塞I/O模型 单线程非阻塞I/O模型 多线程非阻塞I/O模型,Reactor及其...

sea-boat
01/04
0
0
服务器模型——从单线程阻塞到多线程非阻塞(下)

前言的前言 服务器模型涉及到线程模式和IO模式,搞清楚这些就能针对各种场景有的放矢。该系列分成三部分: 单线程/多线程阻塞I/O模型 单线程非阻塞I/O模型 多线程非阻塞I/O模型,Reactor及其...

01/04
0
0
netty学习之Reactor线程模型以及在netty中的应用

1.Reactor单线程模型 Reactor单线程模型就是指所有的IO操作都在同一个NIO线程上面完成的,也就是IO处理线程是单线程的。NIO线程的职责是: (1)作为NIO服务端,接收客户端的TCP连接; (2)...

harries
05/11
0
0
高性能Server---Reactor模型

在这个充斥着云的时代,我们使用的软件可以说99%都是C/S架构的! 你发邮件用的Outlook,Foxmail等 你看视频用的优酷,土豆等 你写文档用的Office365,googleDoc,Evernote等 你浏览网页用的IE,...

tantexian
2017/11/07
0
0
libevent源码深度剖析

原文地址:http://blog.csdn.net/sparkliang/article/details/4957667 libevent源码深度剖析一 ——序幕 张亮 1 前言 Libevent是一个轻量级的开源高性能网络库,使用者众多,研究者更甚,相关...

晨曦之光
2012/03/09
145
0
java十分钟速懂知识点——NIO

一、引子 nio是java的IO框架里边十分重要的一部分内容,其最核心的就是提供了非阻塞IO的处理方式,最典型的应用场景就是处理网络连接。很多同学提起nio都能说起一二,但是细究其背后的原理、...

oO脾气不坏Oo
2015/11/29
389
2
消息中间件—RocketMQ的RPC通信(二)

文章摘要:如何设计RPC通信层模型是任何一款性能强劲的MQ所要重点考虑的问题 在(一)篇中主要介绍了RocketMQ的协议格式,消息编解码,通信方式(同步/异步/单向)、消息发送/接收以及异步回调...

癫狂侠
07/01
0
0
我的开源项目汇总(机器&深度学习、NLP、网络IO、AIML、mysql协议、chatbot)

TextAnalyzer https://github.com/sea-boat/TextAnalyzer A text analyzer which is based on machine learning,statistics and dictionaries that can analyze text. So far, it supports h......

超人汪小建
07/12
0
0
我的2017年文章汇总——Java及中间件篇

2018即将到来,大家看着2017给自己制定的计划有没有感慨?当你觉得过去一年没有什么进步时,那么请行动起来,能开始总是好的。 近期准备把过去一年写的文章按照分类重新整理推送一遍,包括:...

sea-boat
2017/12/29
0
0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

shell中的函数、shell中的数组、告警系统需求分析

shell中的函数 格式: 格式: function f_name() { command } 函数必须要放在最前面 示例1(用来打印参数) 示例2(用于定义加法) 示例3(用于显示IP) shell中的数组 shell中的数组1 定义数...

Zhouliang6
今天
2
0
用 Scikit-Learn 和 Pandas 学习线性回归

      对于想深入了解线性回归的童鞋,这里给出一个完整的例子,详细学完这个例子,对用scikit-learn来运行线性回归,评估模型不会有什么问题了。 1. 获取数据,定义问题     没有...

wangxuwei
今天
1
0
MAC安装MAVEN

一:下载maven压缩包(Zip或tar可选),解压压缩包 二:打开终端输入:vim ~/.bash_profile(如果找不到该文件新建一个:touch ./bash_profile) 三:输入i 四:输入maven环境变量配置 MAVEN_HO...

WALK_MAN
今天
0
0
33.iptables备份与恢复 firewalld的9个zone以及操作 service的操作

10.19 iptables规则备份和恢复 10.20 firewalld的9个zone 10.21 firewalld关于zone的操作 10.22 firewalld关于service的操作 10.19 iptables规则备份和恢复: ~1. 保存和备份iptables规则 ~2...

王鑫linux
今天
2
0
大数据教程(2.11):keeperalived+nginx高可用集群搭建教程

上一章节博主为大家介绍了目前大型互联网项目的系统架构体系,相信大家应该注意到其中很重要的一块知识nginx技术,在本节博主将为大家分享nginx的相关技术以及配置过程。 一、nginx相关概念 ...

em_aaron
今天
1
0
Apache Directory Studio连接Weblogic内置LDAP

OBIEE默认使用Weblogic内置LDAP管理用户及组。 要整理已存在的用户及组,此前办法是导出安全数据,文本编辑器打开认证文件,使用正则表达式获取用户及组的信息。 后来想到直接用Apache Dire...

wffger
今天
2
0
HFS

FS,它是一种上传文件的软件。 专为个人用户所设计的 HTTP 档案系统 - Http File Server,如果您觉得架设 FTP Server 太麻烦,那么这个软件可以提供您更方便的档案传输系统,下载后无须安装,...

garkey
今天
1
0
Java IO类库之BufferedInputStream

一、BufferedInputStream介绍 /** * A <code>BufferedInputStream</code> adds * functionality to another input stream-namely, * the ability to buffer the input and to * sup......

老韭菜
今天
0
0
STM 32 窗口看门狗

http://bbs.elecfans.com/jishu_805708_1_1.html https://blog.csdn.net/a1985831055/article/details/77404131...

whoisliang
昨天
1
0
Dubbo解析(六)-服务调用

当dubbo消费方和提供方都发布和引用完成后,第四步就是消费方调用提供方。 还是以dubbo的DemoService举例 -- 提供方<dubbo:application name="demo-provider"/><dubbo:registry address="z...

青离
昨天
2
0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

返回顶部
顶部