java NIO中的Reactor相关知识汇总 （转）

一、引子

nio是java的IO框架里边十分重要的一部分内容，其最核心的就是提供了非阻塞IO的处理方式，最典型的应用场景就是处理网络连接。很多同学提起nio都能说起一二，但是细究其背后的原理、思想往往就开始背书，说来说去都是那么几句，其中不少人并不见的真的很理解。本人之前就属于此类，看了很多书和博客，但是大多数都只是讲了三件套和怎么使用，很少会很细致的讲背后的思想，那本次我们就来扒一扒吧。     很多博客描述nio都是这么说的：基于Reactor模式实现的多路非阻塞高性能的网络IO。那么我们就从这个定义来分析，其中两个关键点：多路非阻塞和Reactor模式。(本来想把高性能也算进去，但是后来想想这个应该算前两者的结果)下边我们来分别搞懂这两块。

二、网络IO模型

多路非阻塞其实准确的名字叫做IO多路复用模型，其是linux五种网络模型之一，也是当前网络编程最常使用的模型之一。至于详细的介绍请参考博客：高性能IO模型浅析(这个里边只给出了4中，没有信号驱动IO，但讲的很赞，特别是图)，这里仅作简要介绍和对比：

  • 阻塞IO：java中老的bio便是这种模式，在接到事件（数据到达、数据拷贝完成等）前程序需阻塞等待。优点是编码简单，缺点是效率低，处理程序阻塞会导致cpu利用率很低。
  • 非阻塞IO：在未接到事件时处理程序一直主动轮询，这样处理程序无需阻塞，可以在轮询间歇去干别的，但是轮询会造成重复请求，同样浪费资源。以前java中实现的的伪异步模式就是采用这种思想。
  • IO复用模型：增加了对socket的事件监听器(selector)，从而把处理程序和对应的socket事件解耦，所用的socket连接都注册在监听器，在等待阶段只有监听器会阻塞，处理线程从监听器获取事件对socket连接处理即可，而且一个处理线程可以对应多个连接（前两种一般都是一个socket连接起一个线程，这就是为什么叫复用），有点是节省资源，由于处理程序能够被多个连接复用，因此少数的线程就能处理大量连接。缺点同样因为复用，如果是大量费时处理的连接（如大量连接上传大文件），很容易造成线程占满而导致新连接失败。
  • 信号驱动IO模型：在数据准别阶段无需阻塞，只需向系统注册一个信号，在数据准备好后，系统会响应该信号。该模型依赖于系统实现，而且信号通信使用比较麻烦，因此java中未有对应实现。
  • 异步IO：与信号驱动IO很类似，而且在数据拷贝阶段(指数据从系统缓冲区拷贝至程序自己的缓冲区，其他模型改阶段程序都需要阻塞等待)同样可以异步处理。有点不必多说，效率很高，缺点是依赖系统底层实现。目前很多语言都提供该模型的实现，jdk1.7之后同样在concurrent包中提供了。

对比以上五种模型可以知道，IO复用模型从效率和实现成本综合而言目前是比较好的选择，这就是java基于该模型实现nio的根本原因。上边提到了IO复用模型的实现思想，其实这种思想在其他语言中早已实现(如C++中据说流弊哄哄超10w行代码的ACE,自适配通信环境，就采用了该模型)，并且提出了一个叫Reactor的设计模式。

三、Reactor模式

Reactor模式，翻译过来叫做反引器模式，其目的是在事件驱动的应用中，将一个请求的能够分离并且调度给应用程序。我相信大多数人都没看明白前一句的意思（书还是要背的），说白了就是对于一个请求的多个事件（如连接、读写等），经过这种模式的处理，能够区分出来，并且分别交给对应的处理模块处理。废话不多说，来看下一个简图：

可以看到Reactor模式中组件有acceptor、dispatcher和handler(这里只是拿一种实现做个例子，真实的实现各有不同)，其中acceptor中注册了各类事件，当连接有新的事件过来时，其会将事件交给dispatcher进行分发；dispatcher绑定了事件和对应处理程序handler的映射关系，当接到新事件时其会把事件分发到对应handler；而handler负责处理对应事件，这块就是我们的业务层了。

从该模式我们可以发现，对于acceptor、dispatcher我们往往只需要一个线程作为入口即可，因为其并不会有耗时处理，效率很高，而handler则根据需要起几个线程即可（多数时候使用一个线程池实现），这正是IO复用模型期望的效果。

下边我们会介绍NIO是如何实现该模式的，在此之前先介绍一下框架，其实除了NIO之外，基于JVM实现的还有其他Reactor框架，正好最近OSC牵头翻译了对应文档，有兴趣的可以看下：Reactor 指南。

四、NIO

NIO的细节就不多讲了，这里只介绍下三件套：

  • channel：管道，可以看做对流的封装，有点像pipe，不过其是全双工的。其好处是屏蔽了底层细节，不用关心流对应的是文件还是网络，也不用关心连接怎么处理的，而且全双工，不用考虑输入流或输出流，你只用使用buffer对其进行读写就行了。
  • buffer：channel的好基友，底层就是个字节数组，不同的是对其进行了封装，不仅提供了对基本类型的支持，而且内部维持了读写位置(postion、limit、capacity、mark等)，还提供了便捷的方法(clear、flip)。对channel的读写必须通过buffer。
  • selector：这个不多说了，如果前边认真看基本上就明白干啥的，就是Reactor模式中Acceptor的实现。

再来看个简图吧：

基本上和Reactor能对应上，少了个dispatcher，这是由于jdk本身提供的nio比较基本，dispatcher一般都由我们自己实现，而在我理解中，mina、netty这些框架很重要的一方面也是提供了该部分的实现。

五、一个例子

从《netty权威指南》上抄了个例子以及配图，而且代码没有客户端的，大家可以瞄一眼吧（为什么没有？因为已经快一点了，我不想写了......）: 服务器端时序图：

客户端时序图：

服务器端代码：

package com.gj.netty.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.util.Iterator;
import java.util.Set;

/**
 * Created by guojing on 2015/6/7.
 */
public class MultiplexerTimerServer implements Runnable {

    private Selector selector;
    private ServerSocketChannel servChannel;
    private volatile boolean stop;

    public MultiplexerTimerServer(int port) {
        try {
            selector = Selector.open(); //新建多路复用selector
            servChannel = ServerSocketChannel.open();   //新建channel
            servChannel.configureBlocking(false);  //设置非阻塞
            servChannel.socket().bind(new InetSocketAddress(port),1024); //端口、块大小
            servChannel.register(selector, SelectionKey.OP_ACCEPT);
            System.out.println("TimeServer is start, port:" + port);
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    public void run() {
        while (!stop){
            try {
                selector.select(1000);
                Set<SelectionKey> keys = selector.selectedKeys();
                Iterator<SelectionKey> ketIt = keys.iterator();
                SelectionKey key = null;
                while (ketIt.hasNext()){
                    key = ketIt.next();
                    ketIt.remove();
                    //处理对应key事件
                    handler(key);
                }
            } catch (IOException e) {
                e.printStackTrace();
            }

        }
    }

    private void handler(SelectionKey key){
        //根据key去除channel做对应处理
    }
}

http://www.cnblogs.com/good-temper/p/5003892.html

NIO 有一个主要的类Selector，这个类似一个观察者，只要我们把需要探知的socketchannel告诉Selector，我们接着做别的事情，当有事件发生时，他会通知我们，传回一组SelectionKey，我们读取这些Key，就会获得我们刚刚注册过的socketchannel，然后，我们从这个Channel中读取数据，接着我们可以处理这些数据。

反应器模式与观察者模式在某些方面极为相似：
当一个主体发生改变时，所有依属体【在主体中注册的对象】都得到通知。不过，观察者模式与单个事件源关联，而反应器模式则与多个事件源关联 。

一般模型

我们想象以下情形：长途客车在路途上，有人上车有人下车，但是乘客总是希望能够在客车上得到休息。

传统的做法是：每隔一段时间（或每一个站），司机或售票员对每一个乘客询问是否下车。

反应器模式做法是：汽车是乘客访问的主体（Reactor），乘客上车后，到售票员（acceptor）处登记，之后乘客便可以休息睡觉去了，当到达乘客所要到达的目的地后，售票员将其唤醒即可。

Demo:

package com.reactor;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.util.Iterator;
import java.util.Set;

/**
 * 反应器模式
 * 用于解决多用户访问并发问题
 *
 * 举个例子：餐厅服务问题
 *
 * 传统线程池做法：来一个客人(请求)去一个服务员(线程)
 * 反应器模式做法：当客人点菜的时候，服务员就可以去招呼其他客人了，等客人点好了菜，直接招呼一声“服务员”
 *
 * @author linxcool
 */
public class Reactor implements Runnable{
    public final Selector selector;
    public final ServerSocketChannel serverSocketChannel;

    public Reactor(int port) throws IOException{
        selector=Selector.open();
        serverSocketChannel=ServerSocketChannel.open();
        InetSocketAddress inetSocketAddress=new InetSocketAddress(InetAddress.getLocalHost(),port);
        serverSocketChannel.socket().bind(inetSocketAddress);
        serverSocketChannel.configureBlocking(false);

        //向selector注册该channel
        SelectionKey selectionKey=serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

        //利用selectionKey的attache功能绑定Acceptor 如果有事情，触发Acceptor
        selectionKey.attach(new Acceptor(this));
    }

    @Override
    public void run() {
        try {
            while(!Thread.interrupted()){
                selector.select();
                Set<SelectionKey> selectionKeys= selector.selectedKeys();
                Iterator<SelectionKey> it=selectionKeys.iterator();
                //Selector如果发现channel有OP_ACCEPT或READ事件发生，下列遍历就会进行。
                while(it.hasNext()){
                    //来一个事件 第一次触发一个accepter线程
                    //以后触发SocketReadHandler
                    SelectionKey selectionKey=it.next();
                    dispatch(selectionKey);
                    selectionKeys.clear();
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 运行Acceptor或SocketReadHandler
     * @param key
     */
    void dispatch(SelectionKey key) {
        Runnable r = (Runnable)(key.attachment());
        if (r != null){
            r.run();
        }
    }  

}
package com.reactor;

import java.io.IOException;
import java.nio.channels.SocketChannel;

public class Acceptor implements Runnable{
    private Reactor reactor;
    public Acceptor(Reactor reactor){
        this.reactor=reactor;
    }
    @Override
    public void run() {
        try {
            SocketChannel socketChannel=reactor.serverSocketChannel.accept();
            if(socketChannel!=null)//调用Handler来处理channel
                new SocketReadHandler(reactor.selector, socketChannel);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
package com.reactor;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;

public class SocketReadHandler implements Runnable{
    private SocketChannel socketChannel;
    public SocketReadHandler(Selector selector,SocketChannel socketChannel) throws IOException{
        this.socketChannel=socketChannel;
        socketChannel.configureBlocking(false);

        SelectionKey selectionKey=socketChannel.register(selector, 0);

        //将SelectionKey绑定为本Handler 下一步有事件触发时，将调用本类的run方法。
        //参看dispatch(SelectionKey key)
        selectionKey.attach(this);

        //同时将SelectionKey标记为可读，以便读取。
        selectionKey.interestOps(SelectionKey.OP_READ);
        selector.wakeup();
    }

    /**
     * 处理读取数据
     */
    @Override
    public void run() {
        ByteBuffer inputBuffer=ByteBuffer.allocate(1024);
        inputBuffer.clear();
        try {
            socketChannel.read(inputBuffer);
            //激活线程池 处理这些request
            //requestHandle(new Request(socket,btt));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

http://blog.csdn.net/linxcool/article/details/7771952

时间： 2024-10-05 04:27:44

java NIO中的Reactor相关知识汇总 （转）的相关文章

[转帖]xserver相关知识汇总

xserver相关知识汇总 https://blog.csdn.net/QTVLC/article/details/81739984 本文主要是从以下几个方面介绍xorg-xserver 相关的知识 1.linux系统图形界面框架 2.xserver 和x client启动过程 3.图形2d,3d加速原理简介 4.xserver主分支代码解析. 5.xserver,xclient协议简介 6.一个基于Xlib的简单例子解析 7.radeon驱动初始化代码解析. 1.linux图形界面框架 参考至

【干货】样品气体预处理系统相关知识汇总

武汉天禹智控科技有限公司依托多年来从事气体分析行业的经验和强大的技术实力,经过多年研制开发和应用实践,在传感技术方面,成功的研制出拥有自主知识产权的电化学分析仪系列,红外分析仪系列,紫外气体分析仪系列,激光分析仪系列,光声光谱气体分析仪系列,适用于各种工况的工业过程分析系统,同时可以根据客户需求进行气体分析仪个性定制,产品广泛应用于环保.冶金.石化.化工.能源.食品.农业.交通.水利.建筑.制药.酿造及科学研究等众多行业,并且得到用户的一直好评.现针对样品气体预处理系统相关知识汇总分享,供大家使

Java NIO中的缓冲区Buffer（一）缓冲区基础

什么是缓冲区(Buffer) 定义 简单地说就是一块存储区域,哈哈哈,可能太简单了,或者可以换种说法,从代码的角度来讲(可以查看JDK中Buffer.ByteBuffer.DoubleBuffer等的源码),Buffer类内部其实就是一个基本数据类型的数组,以及对这个缓冲数组的各种操作: 常见的缓冲区如ByteBuffer.IntBuffer.DoubleBuffer...内部对应的数组依次是byte.int.double... 与通道的关系 在Java NIO中,缓冲区主要是跟通道(Chann

java NIO中的buffer和channel

缓冲区(Buffer):一,在 Java NIO 中负责数据的存取.缓冲区就是数组.用于存储不同数据类型的数据 根据数据类型不同(boolean 除外),提供了相应类型的缓冲区:ByteBufferCharBufferShortBufferIntBufferLongBufferFloatBufferDoubleBuffer 上述缓冲区的管理方式几乎一致,通过 allocate() 获取缓冲区 二.缓冲区存取数据的两个核心方法:put() : 存入数据到缓冲区中get() : 获取缓冲区中的数据

Java NIO中的Glob模式详解

Java NIO中的Glob模式详解 作者:chszs,转载需注明.博客主页:http://blog.csdn.net/chszs 一.什么是Glob? 在编程设计中,Glob是一种模式,它使用通配符来指定文件名.例如:.java就是一个简单的Glob,它指定了所有扩展名为"java"的文件.Glob模式中广泛使用了两个通配符""和"?".其中星号表示"任意的字符或字符组成字符串",而问号则表示"任意单个字符&quo

Java NIO中的Buffer 详解

Java NIO中的Buffer用于和NIO通道进行交互.如你所知,数据是从通道读入缓冲区,从缓冲区写入到通道中的.缓冲区本质上是一块可以写入数据,然后可以从中读取数据的内存.这块内存被包装成NIO Buffer对象,并提供了一组方法,用来方便的访问该块内存. Buffer的基本用法使用Buffer读写数据一般遵循以下四个步骤: 写入数据到Buffer    调用flip()方法    从Buffer中读取数据    调用clear()方法或者compact()方法 当向buffer写入数据时,

Java NIO 与 基于reactor设计模式的事件处理模型

Java NIO非堵塞应用通常适用用在I/O读写等方面,我们知道,系统运行的性能瓶颈通常在I/O读写,包括对端口和文件的操作上,过去,在打开一个I/O通道后,read()将一直等待在端口一边读取字节内容,如果没有内容进来,read()也是傻傻的等,这会影响我们程序继续做其他事情,那么改进做法就是开设线程,让线程去等待,但是这样做也是相当耗费资源的. Java NIO非堵塞技术实际是采取Reactor模式,或者说是Observer模式为我们监察I/O端口,如果有内容进来,会自动通知我们,这样,我们

Logback相关知识汇总

例如:%-4relative 表示,将输出从程序启动到创建日志记录的时间 进行左对齐 且最小宽度为4格式修饰符,与转换符共同使用:可选的格式修饰符位于“%”和转换符之间.第一个可选修饰符是左对齐 标志,符号是减号“-”:接着是可选的最小宽度 修饰符,用十进制数表示.如果字符小于最小宽度,则左填充或右填充,默认是左填充(即右对齐),填充符为空格.如果字符大于最小宽度,字符永远不会被截断.最大宽度 修饰符,符号是点号"."后面加十进制数.如果字符大于最大宽度,则从前面截断.点符号“.”后面

Java工程师应该掌握的相关知识(基础篇01)

自己并不是 CS 科班出身,学习 Java 的时间也不长,但自己比较喜欢这行业.所以想写一些文章记录一些自己想去详细了解的知识,并且希望能分享出来大家进行讨论,一起进步~ 我比较喜欢问题驱动学习的形式,所以经常会看一些大公司的面试题了解自己的不足,下面有五道面试题与大家讨论.下面讨论的问题出自 —— 你应该知道的JAVA面试题 1. Java 线程的状态 这算是一个比较常见的问题了,经常在不同的面经里看见,但经常又记了大概就忘记. Java 线程在某个时刻只能处于以下六种状态 1. 新建(NEW