2.NIO入门

本章着重掌握如下2个知识点：

• 传统的同步阻塞式I/O编程
• 基于NIO的非阻塞编程
• 基于NIO2.0的异步非阻塞（AIO）编程
• 为什么要使用NIO编程

---

传统的BIO编程

介绍: 网络编程的基本模型是Client/Server模型，也就是两个进程之间进行相互通信，其中服务端提供位置信息（绑定的IP地址和监听端口），客服端通过连接操作向服务端监听的地址发起连接请求，通过三次握手建立连接，如果连接成功，双方就可以通过网络套接字（Socket）进行通信。在基于传统同步阻塞模型开发中，ServerSocket负责绑定IP地址，启动监听端口；Socket负责发起连接操作。连接成功之后，双方通过输入和输出流进行同步阻塞式通信。

BIO通信模型图

我们根据下图所示的通信模型图来熟悉BIO的服务端通信模型：采用BIO通信模型的服务端，通常由一个独立的Acceptor线程负责监听客户端的连接，它接收到客户端连接请求之后为每个客户端创建一个新的线程进行链路处理，处理完成之后，通过输出流返回应答给客户端，线程销毁，这就是典型的一请求一应答通信模型。

该模型最大的问题就是缺乏弹性伸缩能力，当客户端并发访问量增加后，服务端的线程个数和客户端并发访问数呈1:1的正比关系，由于线程是Java虚拟机非常宝贵的系统资源，当线程数膨胀之后，系统的性能将急剧下降，随着并发访问量的继续增大，系统会发生线程堆栈溢出、创建新线程失败等问题，并最终导致进程宕机或者僵死，不能对外提供服务。

看个例子（同步阻塞是I/O创建的TimeServer）:

服务端:

package netty;

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;

/**
 * Created by yongshan_ning on 2017/5/17.
 */
public class TimeServer {
    public static void main(String[] args) throws IOException{
        int port = 8080;
        if(args != null && args.length > 0){
            try{
                port = Integer.valueOf(args[0]);
            }catch (NumberFormatException e){
                e.printStackTrace();
            }
        }
        ServerSocket server = null;
        try{
            server = new ServerSocket(port);
            System.out.println("The time server is start in port :" + port);
            Socket socket = null;
            while (true){
                socket = server.accept();
                new Thread(new TimeServerHandler(socket)).start();
            }
        }finally {
            if(server != null){
                System.out.println("The time server close");
                server.close();
                server = null;
            }
        }
    }
}

package netty;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;

/**
 * Created by yongshan_ning on 2017/5/17.
 */
public class TimeServerHandler implements Runnable{
    private Socket socket;

    public TimeServerHandler(Socket socket){
        this.socket = socket;
    }

    @Override
    public void run(){
        BufferedReader in = null;
        PrintWriter out = null;
        try{
            in = new BufferedReader(new InputStreamReader(this.socket.getInputStream()));
            out = new PrintWriter(this.socket.getOutputStream(), true);
            String currentTime = null;
            String body = null;
            while(true){
                body = in.readLine();
                if(body == null){
                    break;
                }
                System.out.println("The time server receive order :" + body);
                currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ?
                        new java.util.Date(System.currentTimeMillis()).toString() : "BAD ORDER";
                out.println(currentTime);
            }
        }catch (Exception e){
            if(in != null){
                try{
                    in.close();
                }catch (IOException e1){
                    e1.printStackTrace();
                }
            }
            if(out != null){
                out.close();
                out = null;
            }
            if(this.socket != null){
                try{
                    this.socket.close();
                }catch (IOException e1){
                    e1.printStackTrace();
                }
                this.socket = null;
            }
        }
    }
}

分析：TimeServer根据传入的参数设置监听端口，如果没有入参，使用默认值8080。通过构造函数创建ServerSocket，如果端口合法且没有被占用，服务端监听成功。通过一个无限循环来监听客户端的连接，如果没有客户端接入，则主线程阻塞在ServerSocket的accept操作上。启动TimeServer，通过JvisualVM打印线程堆栈，我们可以发现主程序确实阻塞在accept操作上，如下图所示：

当有新的客户端接入的时候，以Socket为参数构造TimeServerHandler对象，TimeServerHandler是一个Runnable，使用它为构造函数的参数创建一个新的客户端线程处理这条Socket链路。

客户端:

package netty;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;

/**
 * Created by yongshan_ning on 2017/5/19.
 */
public class TimeClient {
    public static void main(String[] args){
        int port = 8080;
        if(args != null && args.length > 0){
            try{
                port = Integer.valueOf(args[0]);
            }catch (NumberFormatException e){
                e.printStackTrace();
            }
        }
        Socket socket = null;
        BufferedReader in = null;
        PrintWriter out = null;
        try{
            socket  = new Socket("127.0.0.1", port);
            in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
            out = new PrintWriter(socket.getOutputStream(), true);
            out.println("Query TIME ORDER");
            System.out.println("Send order 2 server succeed.");
            String resp = in.readLine();
            System.out.println("Now is :" + resp);
        }catch (Exception e){

        }finally {
            if(out != null){
                out.close();
                out = null;
            }

            if(in != null){
                try{
                    in.close();
                }catch (IOException e){
                    e.printStackTrace();
                }
                in = null;
            }
            if(socket != null){
                try{
                    socket.close();
                }catch (IOException e){
                    e.printStackTrace();
                }
                socket = null;
            }
        }

    }
}

分别执行服务端和客户端，结果分别如下:

分析：通过实验发现，BIO主要的问题在于每当有一个新的客户端请求接入时，服务端必须创建一个新的线程处理新接入的客户端链路，一个线程只能处理一个客户端连接。在高性能服务器应用领域，往往需要面向成千上万个客户端的并发连接，这种模型显然无法满足高性能、高并发接入的场景。

伪异步I/O编程

介绍: 采用线程池和任务队列可以实现一种叫做伪异步的I/O通信框架，它的模型图如下图所示。

当有新的客户端接入时，将客户端的Socket封装成一个Task（该任务实现java.lang.Runnable接口）投递到后端的线程池中进行处理，JDK的线程池维护一个消息队列和N个活跃线程，对消息队列中的任务进行处理。由于线程池可以设置消息队列的大小和最大线程数，因此，它的资源占用是可控的，无论多少个客户端并发访问，都不会导致资源的耗尽和宕机。接下来我们通过一个例子来说明此模型的弊端。

看个例子（伪异步I/O创建的TimeServer）:

服务端:

package netty.weiIO;

import netty.TimeServerHandler;

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;

/**
 * Created by yongshan_ning on 2017/5/22.
 */
public class TimeServer {
    public static void main(String[] args) throws IOException{
        int port = 8080;
        if(args != null && args.length > 0){
            try{
                port = Integer.valueOf(args[0]);
            }catch (NumberFormatException e){
                //采用默认值
            }
        }
        ServerSocket server = null;
        try{
            server = new ServerSocket(port);
            System.out.println("The time server is start in port :" + port);
            Socket socket = null;
            TimeServerHandlerExecutePool singleExecutor = new TimeServerHandlerExecutePool(50, 10000); // 创建I/O任务线程池
            while (true){
                socket = server.accept();
                singleExecutor.execute(new TimeServerHandler(socket));
            }
        }finally {
            if(server != null){
                System.out.println("The time server close");
                server.close();
                server = null;
            }
        }
    }
}

package netty.weiIO;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Created by yongshan_ning on 2017/5/22.
 */
public class TimeServerHandlerExecutePool
{
    private ExecutorService executor;

    public TimeServerHandlerExecutePool(int maxPoolSize, int queueSize){
        executor = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(), maxPoolSize, 120L, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(queueSize));
    }

    public void execute(Runnable task){
        executor.execute(task);
    }
}

客户端:

package netty;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;

/**
 * Created by yongshan_ning on 2017/5/19.
 */
public class TimeClient {
    public static void main(String[] args){
        int port = 8080;
        if(args != null && args.length > 0){
            try{
                port = Integer.valueOf(args[0]);
            }catch (NumberFormatException e){
                e.printStackTrace();
            }
        }
        Socket socket = null;
        BufferedReader in = null;
        PrintWriter out = null;
        try{
            socket  = new Socket("127.0.0.1", port);
            in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
            out = new PrintWriter(socket.getOutputStream(), true);
            out.println("Query TIME ORDER");
            System.out.println("Send order 2 server succeed.");
            String resp = in.readLine();
            System.out.println("Now is :" + resp);
        }catch (Exception e){

        }finally {
            if(out != null){
                out.close();
                out = null;
            }

            if(in != null){
                try{
                    in.close();
                }catch (IOException e){
                    e.printStackTrace();
                }
                in = null;
            }
            if(socket != null){
                try{
                    socket.close();
                }catch (IOException e){
                    e.printStackTrace();
                }
                socket = null;
            }
        }

    }
}

分别执行服务端和客户端，结果分别如下:

分析 : 由于线程池和消息队列都是有界的，因此，无论客户端并发连接数多大，它都不会导致线程个数过于膨胀或者内存溢出，相比于传统的一连接一线程模型，是一种改良。伪异步I/O通信框架采用了线程池实现，因此避免了为每个请求都创建一个独立线程照成的线程资源耗尽问题。但是由于它底层的通信依然采用同步阻塞模型，因此无法从根本上解决问题。

伪异步I/O弊端分析

介绍:要对伪异步I/O的弊端进行深入分析，首先来分下一下Java同步I/O的API说明。

输入流 :

首先从输入流进行分析。从上图蓝色背景的文字部分的API说明（InputStream.java），当对Socket的输入流进行读取操作的时候，它会一直阻塞下去，直到发生如下三种事件：

• 有数据可读；
• 可用数据已经读取完毕；
• 发生空指针或者I/O异常。

这意味着当对方发送请求或者应答消息比较缓慢，或者网络传输较慢时，读取输入流一方的通信线程将被长时间阻塞，如果对方要60S才能够将数据发送完成，读取一方的I/O线程也将会被同步阻塞60S，在此期间，其他接入消息只能在消息队列中排队。

输出流 :

当调用OutputStream的write方法写输入流的时候，它将会被阻塞，直到所有要发送的字节全部写入完毕，或者发生异常。学习过TCP/IP相关知识的人都知道，当消息的接收方处理缓慢的时候，将不能及时地从TCP缓冲区读取数据，这将会导致发送方的TCP的window sise不断减小，直到为0，双方处于Keep-Alive状态，消息发送方将不能再向TCP缓冲区写入消息，这时如果采用的是同步阻塞I/O，write操作将会被无限期阻塞，直到TCP的window size大于0或者发生I/O异常。

总结：通过对输入和输出流的API文档进行分析，我们了解到读和写操作都是同步阻塞的，阻塞的时间取决于对方I/O线程的处理速度和网络I/O的传输速度。本质上来讲，我们无法保证生产环境的网络状况和对端的应用程序足够快，如果我们的应用程序依赖对方的处理速度，它的可靠性就会非常差。

伪异步I/O实际上仅仅是对之前I/O线程模型的一个简单优化，它无法从根本上解决同步I/O导致的通信线程阻塞问题。下面我们就简单分析下通信对方返回应答时间过长会引起的级联故障：
- 服务端处理缓慢，返回应答消息耗费60s，平时只需要10ms。

• 采用伪异步I/O的线程正在读取故障服务节点的响应，由于读取输入流是阻塞的，它将会被同步阻塞60s。

• 假如所有的可用线程都被故障服务器阻塞，那后续所有的I/O消息都将在队列中排队。

• 由于线程池采用阻塞队列实现，当队列积满之后，后续入队列的操作将被阻塞。

• 由于前端只有一个Accptor线程接收客户端接入，它被阻塞在线程池的同步阻塞队列之后，新的客户端请求消息将被拒绝，客户端会发生大量的连接超时。

• 由于几乎所有的连接都超时，调用者会认为系统已经崩溃，无法接收新的请求消息。

NIO编程

介绍: NIO到底是什么的简称？有人称之为NEW I/O，原因在于它相对于之前的I/O类库是新增的。这时它的官方叫法。但是，由于之前老的I/O类库是阻塞I/O，New I/O类库的目标就是要让Java支持非阻塞I/O，所以，更多的人喜欢称之为非阻塞I/O（Non-block I/O）。由于非阻塞I/O更能体现NIO的特点。

与Socket类和ServerSocket类相对应，NIO也提供了SocketChannel和ServerSocketChannel两种不同的套接字通道实现。这两种新增的通道都支持阻塞和非阻塞两种模式。阻塞模式使用非常简单，但是性能和可靠性都不好，非阻塞模式则正好相反。一般来说，低负载、低并发的应用程序可以选择同步阻塞I/O一降低编程复杂度；对于高负载、高并发的网络应用，需要使用NIO的非阻塞模式进行开发。

NIO类库介绍

介绍：新的输入/输出（NIO）库是在JDK1.4中引入的。NIO弥补了原来同步阻塞I/O的不足，它在标准Java代码中提供了高速的、面向块的I/O。

1. 缓冲区Buffer

介绍：Buffer是一个对象，它包含一些要写入或者要读出的数据。在NIO类库中加入Buffer对象，体现了新库与原I/O的一个重要区别。在面向流的I/O中，可以将数据直接写入或者将数据直接读到Stream对象中。

在NIO库中，所有数据都是用缓冲区处理的。在读取数据时，它是直接读到缓冲区中的；在写入数据时，写入到缓冲区中。任何时候访问NIO中的数据，都是通过缓冲区进行操作。

缓冲区实质上是一个数组。通常它是一个字节数组（ByteBuffer），也可以使用其他种类的数组。但是一个缓冲区不仅仅是一个数组，缓冲区提供了对数据的结构化访问以及维护读写位置（limit）等信息。

最常用的缓冲区是ByteBuffer，一个ByteBuffer提供了一组功能用于操作byte数组。除了ByteBuffer，还有其他的一些缓冲区，事实上，每一种Java基本类型（除了Boolean类型）都对应有一种缓冲区，具体如下：

• ByteBuffer：字节缓冲区

• CharBuffer：字符缓冲区

• ShortBuffer：短整型缓冲区

• IntBuffer：整型缓冲区

• LongBuffer：长整型缓冲区

• FloatBuffer：浮点型缓冲区

• DoubleBuffer：双精度浮点型缓冲区

每一个Buffer类都是Buffer接口的一个子实例。除了ByteBuffer，每一个Buffer类都有完全一样的操作，只是它们所处理的数据类型不一样。因为大多数标准I/O操作都使用ByteBuffer，所以它在具有一般缓冲区的操作之外还提供了一些特有的操作，以方便网络读写。

2. 通道Channel

介绍: Channel是一个通道，它就像自来水管一样，网络数据通过Channel读取和写入。通道与流的不同之处在于通道是双向的，流只是在一个方向上移动（一个流必须是InputStream或者OutputStream的子类），而通道可以用于读、写或者二者同时进行。

因为Channel是全双工的，所以它可以比流更好地映射底层操作系统的API。特别是在UNIX网络编程模型中，底层操作系统的通道都是全双工的，同时支持读写操作。Channel的类图集成关系如下图：

自顶向下看，前三层主要是Channel接口，用于定义它的功能，后面是一些具体的功能类（抽象类）。从类图可以看出，实际上Channel可以分为两大类：用于网络读写的SelectableChannel和用于文件操作的FileChannel。之后涉及的ServerSocketChannel和SocketChannel都是SelectableChannel的子类。

2. 多路复用器Selector

介绍: 多路复用器Selector是Java NIO编程的基础，熟练地掌握Selector对于NIO编程至关重要。多路复用器提供选择已经就绪的任务的能力。简单地讲，Selector会不断地轮询注册在其上的Channel，如果某个Channel上面发生读或者写时间，这个Channel就处于就绪状态，会被Selector轮询出来，然后通过SelectionKey可以获取就绪Channel的集合，进行后续的I/O操作。

一个多路复用器Selector可以同时轮询多个Channel，由于JDK使用而来epoll（）代替传统的select实现，所以它并没有最大连接句柄1024/2048的限制。这也就以为着只需要一个线程负责Selector的轮询，就可以接入成千上万的客户端，这确实是个非常巨大的进步。

NIO服务端序列图

NIO服务端通信序列图如下图所示：

下面将以上图为基础，对NIO服务端的主要创建过程进行讲解和说明：

步骤一：打开SocketChannel，绑定客户端本地地址（可选，默认系统会随机分配一个可用的本地地址），实例代码如下：

SocketChannel client = SocketChannel.open();

步骤二：设置SocketChannel连接为非阻塞模式，同时设置客户端连接的TCP参数，实例代码如下：

acceptorSvr.socket.bind(new InetSocketAddress(InetAddress.getByName("ip"), port));
acceptorSvr.configureBlocking(false);

步骤三：异步连接服务端，实例代码如下：

Selector selector = Selector.open();
New Thread(new ReactorTask()).start();

步骤四：判断是否连接成功，如果连接成功，则直接注册读状态位到多路复用器中，如果当前没有连接成功（异步连接，返回false，说明客户端已经发送sync包，服务端没有返回ack包，物理链路还没有建立），示例代码如下：

SelectionKey key = acceptorSvr.register(selector, SelectionKey.OP_ACCEPT, ioHandler);

步骤五：向Reactor线程的多路复用器注册OP_CONNECT状态位，监听服务端的TCP ACK应答，示例代码如下：

int num = selector.select();
Set selectedKyes = selector.selectedKeys();
Iterator it = selectedKeys.iterator();
while(it.hasNext()){
    Selectionkey key = (SelectionKey)it.next();
    //...deal with I/O event ...
}

步骤六：创建Reactor线程，创建多路复用器并启动线程，示例代码如下：

SocketChannel channel = srvChannel.accept();

步骤七：，示例代码如下：

channel.configureBlocking(false);
channel.socket().setReuseAddress(true);

步骤八：将新接入的客户端连接注册到Reactor线程的多路复用器上，监听读操作，读取客户端发送的网络消息，示例代码如下：

SelectionKey key = socketChannel.register(selector, SelectionKey.OP_READ, ioHandler);

步骤九：异步读取客户端请求消息到缓冲区，示例代码如下：

int readNumber = channel.read(receivedBuffer);

步骤十：对ByteBuffer进行编解码，如果有半包消息指针reset，继续读取后续的报文，将解码成功的消息封装成Task，投递到业务线程池中，进行业务逻辑编码，示例代码如下：

Object message = null;
while(buffer.hasRemain())
{
    byteBuffer.mark();
    Object message = decode(byteBuffer)
    if(message == null)
    {
        byteBuffer.reset();
        break;
    }
    messageList.add(message);
}
if(!byteBuffer.hasRemain())
{
    byteBuffer.clear();
}else{
    byteBuffer.compact();
}
if(messageList != null & !messageList.isEmpty())
{
    for(Object messageE : messageList)
    {
        handlerTask(messageE);
    }   
}

步骤十一：将POJO对象encode成ByteBuffer，调用SocketChannel的异步write接口将，将消息异步发送给客户端，示例代码如下：

socketChannel.write(buffer);

注意：如果发送区TCP缓冲区满，会导致写半包，此时，需要注册监听写操作位，循环写，知道整包消息写入TCP缓冲区。

NIO创建的TimeServer

package netty.NIO;

import java.io.IOException;

/**
 * Created by yongshan_ning on 2017/5/25.
 */
public class TimeServer {
    public static void main(String[] args) throws IOException{
        int port = 8080;
        if(args != null && args.length > 0){
            try{
                port = Integer.valueOf(args[0]);
            }catch (NumberFormatException e){

            }
        }
        MultiplexerTimeServer timeServer = new MultiplexerTimeServer(port);
        new Thread(timeServer, "NIO-MultiplexerTimerServer-001").start();
    }

}

package netty.NIO;


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

/**
 * Created by yongshan_ning on 2017/5/25.
 */
public class MultiplexerTimeServer implements Runnable{
    private Selector selector;

    private ServerSocketChannel servChannel;

    private volatile boolean stop;

    /**
     * 初始化多路复用器，绑定监听端口
     */
    public MultiplexerTimeServer(int port){
        try{
            selector = Selector.open();
            servChannel = ServerSocketChannel.open();
            //设为异步非阻塞模式
            servChannel.configureBlocking(false);
            servChannel.socket().bind(new InetSocketAddress(port), 1024);
            //监听SelectionKey.OP_ACCEPT操作位
            servChannel.register(selector, SelectionKey.OP_ACCEPT);
            System.out.println("The time server is start in port : " + port );
        }catch (IOException e){
            e.printStackTrace();
            System.exit(1);
        }
    }

    public void stop(){
        this.stop = true;
    }

    @Override
    public void run(){
        while (!stop){
            try{
                //selector休眠时间为1s, 无论是否有读写事件发生，selector每隔1s都被唤醒一次
               selector.select(1000);
               Set<SelectionKey> selectionKeySet = selector.selectedKeys();
               Iterator<SelectionKey> it = selectionKeySet.iterator();
               SelectionKey key = null;
               while (it.hasNext()){
                   key = it.next();
                   it.remove();
                   try{
                       handleInput(key);
                   }catch (Exception e){
                       if(key != null){
                           key.cancel();
                           if(key.channel() != null){
                               key.channel().close();
                           }
                       }
                   }
               }
            }catch (Throwable t){
                t.printStackTrace();
            }
        }
        //多路复用器关闭以后所有注册在上面的Channel和Pipe等资源都会被自动去注册并关闭，所以不需要重复释放资源
        if(selector != null){
            try{
                selector.close();
            }catch (IOException e){
                e.printStackTrace();
            }
        }
    }

    private void handleInput(SelectionKey key) throws IOException{
        if(key.isValid()){
            //处理新接入的请求消息
            if(key.isAcceptable()){
                //Accept the new connection
                ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
                SocketChannel sc = ssc.accept();
                sc.configureBlocking(false);
                //Add the new connection to the selector
                sc.register(selector, SelectionKey.OP_READ);
            }

            if(key.isReadable()){
                //Read the data
                SocketChannel sc = (SocketChannel) key.channel();
                ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                int readBytes = sc.read(readBuffer);
                if(readBytes > 0){
                    //将缓冲区当前的limit设置为position, position设置为0,用于后续对缓冲区的读取操作
                    readBuffer.flip();
                    byte[] bytes = new byte[readBuffer.remaining()];
                    readBuffer.get(bytes);
                    String body = new String(bytes, "UTF-8");
                    System.out.println("The time Server receive order :" + body);
                    String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body)?new java.util.Date(System.currentTimeMillis()).toString() : "BAD ORDER";
                    doWrite(sc, currentTime);
                }else if(readBytes < 0){
                    //对端链路关闭
                    key.cancel();;
                    sc.close();
                }else{

                }
            }
        }
    }
    //由于SocketChannel是异步非阻塞的，它并不保证一次能够把需要发送的字节数组发送完，此时会出现"写半包"问题，
    //解决办法：需要注册写操作，不断轮询Selector将没有发送完的ByteBuffer发送完毕，然后可以通过ByteBuffer的hasRemain()方法判断消息是否发送完成
    private void doWrite(SocketChannel channel, String response) throws IOException{
        if(response != null && response.trim().length() > 0){
            byte[] bytes = response.getBytes();
            ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
            writeBuffer.put(bytes);
            writeBuffer.flip();
            channel.write(writeBuffer);
        }
    }
}

NIO客户端序列图

NIO客户端通信序列图如下图所示：

下面将以上图为基础，对NIO客户端的主要创建过程进行讲解和说明：

步骤一：打开SocketChannel，绑定客户端本地地址（可选，默认系统会随机分配一个可用的本地地址），示例代码如下：

 SocketChannel clientChannel = SocketChannel.open();

步骤二：绑定监听端口，设置连接为非阻塞模式，实例代码如下：

clientChannel.configureBlocking(false);
socket.setReuseAddress(true);
socket.setReceiveBufferSize(BUFFER_SIZE);
socket.setSendBufferSize(BUFFER_SIZE);

步骤三：异步连接服务端，实例代码如下：

boolean connected = clientChannel.connect(new InetSocketAddress("ip", port));

步骤四：判断是否连接成功，如果连接成功，则直接注册读状态位到多路复用器中，如果当前没有连接成功（异步连接，返回false，说明客户端已经发送sync包，服务端没有返回ack包，物理链路还没有建立），示例代码如下：

if(connected)
{
    clientChannel.register(selector, SelectionKey.OP_READ, ioHandler);
}
else
{
    clientChannel.register(selector, SelectionKey.OP_CONNECT, ioHandler);
}

步骤五：向Reactor线程的多路复用器注册OP_CONNECT状态位，监听服务端的TCP ACK应答，示例代码如下：

clientChannel.register(selector, SelectionKey.OP_CONNECT, ioHandler);

步骤六：创建Reactor线程，创建多路复用器并启动线程，示例代码如下：

Selector selector = Selector.open();
New Thread(new ReactorTask()).start();

步骤七：多路复用器在线程run方法的无限循环体内轮询准备就绪的Key，示例代码如下：

int num = selector.select();
Set selectedKeys = selector.selectedKeys();
Iterator it = selectedKeys.iterator();
while(it.hasNext()){
    SelectionKey key = (SelectionKey)it.next();
    //... deal with I/O event ...
}

步骤八：接收connect事件进行处理，示例代码如下：

if(key.isConnectable())
//handlerConnect();

步骤九：判断连接结果，如果连接成功，注册读事件到多路复用器，示例代码如下：

if(channel.finishConnect())
    registerRead();

步骤十：注册读事件到多路复用器，示例代码如下：

clientChannel.register(selector, SelectionKey.OP_READ, ioHandler);

步骤十一：异步读客户端请求消息到缓冲区，示例代码如下：

int readNumber = channel.read(receiveBuffer);

步骤十二：对ByteBuffer进行编解码，如果有半包消息接收缓冲区Reset，继续读取后续的报文，讲解码成功的消息封装成Task，投递到业务线程池中，进行业务逻辑编排。示例代码如下：

Object message = null;
while(buffer.hasRemain())
{
    byteBuffer.mark();
    Object message = decode(byteBuffer);
    if(message == null){
        byteBuffer.reset();
        break;
    }
    messageList(meesage);
    if(!byteBuffer.hasRemain()){
        byteBuffer.clear(); 
    }else{
        byteBuffer.compact();
    }
    if(messageList != null & !messageList.isEmpty())
    {
        for(Object messageE : messageList){
            handlerTask(messageE);
        }
    }
}

步骤十三：将POJO对象encode成ByteBuffer，调用SocketChannel的异步write接口，将消息异步发送给客户端。示例代码如下：

socketChannel.write(buffer);

NIO创建的TimeClient

package netty.NIO;

import java.io.IOException;

/**
 * Created by yongshan_ning on 2017/6/2.
 */
public class TimeClient {
    public static void main(String[] args) throws IOException {
        int port = 8080;
        if(args != null && args.length > 0){
            try{
                port = Integer.valueOf(args[0]);
            }catch (NumberFormatException e){

            }
        }
        new Thread(new TimeClientHandle("127.0.0.1", port), "NIO-MultiplexerTimerServer-001").start();
    }
}

package netty.NIO;

import sun.plugin2.os.windows.SECURITY_ATTRIBUTES;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.security.InvalidParameterException;
import java.util.Iterator;
import java.util.Set;

/**
 * Created by yongshan_ning on 2017/6/2.
 */
public class TimeClientHandle implements Runnable{
    private String host;
    private int port;
    private Selector selector;
    private SocketChannel socketChannel;
    private volatile boolean stop;

    /*
       初始化NIO的多路复用器和SocketChannel对象，需要注意的是，创建
       SocketChannel之后，需要将其设置为异步非阻塞模式。
     */
    public TimeClientHandle(String host, int port){
        this.host = host == null ? "127.0.0.1" : host;
        this.port = port;
        try{
            selector = Selector.open();
            socketChannel = SocketChannel.open();
            socketChannel.configureBlocking(false);
        }catch (IOException e){
            e.printStackTrace();
            System.exit(1);
        }
    }

    @Override
    public void run(){
        try{
            doConnect();
        }catch (IOException e){
            e.printStackTrace();
            System.exit(1);
        }
        while (!stop){
            try{
                selector.select(1000);
                Set<SelectionKey> selectedKeys = selector.selectedKeys();
                Iterator<SelectionKey> it = selectedKeys.iterator();
                SelectionKey key = null;
                while (it.hasNext()){
                    key = it.next();
                    it.remove();
                    try{
                        handleInput(key);
                    }catch (Exception e){
                        if(key != null){
                            key.cancel();
                            if(key.channel() != null){
                                key.channel().close();
                            }
                        }
                    }
                }
            }catch (Exception e){
                e.printStackTrace();
                System.exit(1);
            }
        }

        //多路复用器关闭后，所有注册在上面的Channel和pipe等资源都会被自动注册并关闭。所以不需要重复释放资源
        if(selector != null){
            try{
                selector.close();
            }catch (IOException e){
                e.printStackTrace();
            }
        }
    }

    private void handleInput(SelectionKey key) throws IOException{
        if(key.isValid()){
            //判断是否连接成功
            SocketChannel sc = (SocketChannel) key.channel();
            if(key.isConnectable()){
                if(sc.finishConnect()){
                    sc.register(selector, SelectionKey.OP_READ);
                    doWrite(sc);
                }else{
                    System.exit(1); //连接失败， 进程退出
                }
            }

            if(key.isReadable()){
                ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                int readBytes = sc.read(readBuffer);
                if(readBytes > 0){
                    readBuffer.flip();
                    byte[] bytes = new byte[readBuffer.remaining()];
                    readBuffer.get(bytes);
                    String body = new String(bytes, "UTF-8");
                    System.out.println("Now is :" + body);
                    this.stop = true;
                }else if(readBytes < 0){
                    //对端链路关闭
                    key.cancel();
                    sc.close();;
                }else{
                    //读到0字节，忽视
                }
            }
        }
    }

    private void doConnect() throws IOException{
        //如果直接连接成功，则注册到多路复用器上，发送请求消息，读应答
        if(socketChannel.connect(new InetSocketAddress(host, port))){
            socketChannel.register(selector, SelectionKey.OP_READ);
            doWrite(socketChannel);
        }else {
            socketChannel.register(selector, SelectionKey.OP_CONNECT);
        }
    }

    private void doWrite(SocketChannel sc) throws IOException{
        byte[] req = "QUERY TIME ORDER".getBytes();
        ByteBuffer wrireBuffer = ByteBuffer.allocate(req.length);
        wrireBuffer.put(req);
        wrireBuffer.flip();
        sc.write(wrireBuffer);
        if(!wrireBuffer.hasRemaining()){
            System.out.println("Send order 2 server succeed");
        }
    }

}

结果：

使用NIO编程的优点总结如下：

（1）客户端发起的连接操作时异步的，可以通过在多路复用器注册OP_CONNECT等待后续结果，不需要像之前的客户端那样呗同步阻塞。

（2）SocketChannel的读写操作都是异步的，如果没有可读写的数据它不会同步等待，直接返回，这样I/O通信线程就可以处理其他的链路，不需要同步等待这个链路可用。

（3）线程模型的优化：由于JDK的Selector在Linux等主流操作系统上通过epoll实现，它没有连接句柄数的限制（只受限于操作系统的最大句柄数或者对单个进程的句柄限制），这意味着一个Selector线程可以同时处理成千上万个客户端连接，而且性能不会随着客户端的增加而线性下降。因此，它非常适合做高性能、高负载的网络服务器。

AIO编程

介绍: NIO2.0引入了新的异步通道的概念，并提供了异步文件通道和异步套接字通道的实现。异步通道提供一下两种方式获取操作结果。

1. 通过java.util.concurrent.Future类来表示异步操作的结果；
2. 在执行异步操作的时候传入一个java.nio.channels。

CompletionHandler接口的实现类作为操作完成的回调。

NIO2.0的异步套接字通道是真正的异步非阻塞I/O，对应于UNIX网络编程中的事件驱动I/O（AIO）。它不需要通过多路复用器（Selector）对注册的通道进行轮询操作即可实现异步读写，从而简化了NIO的编程模型。

AIO创建的TimeServer

/**
 * Created by ning on 2017/6/10.
 */
public class TimeServer {
    public static void main(String[] args){
        int port = 8080;
        if(args != null && args.length > 0){
            try{
                port = Integer.valueOf(args[0]);
            }catch (NumberFormatException e){
                //采用默认值
            }
        }
        AsyncTimeServerHandler timeServer = new AsyncTimeServerHandler(port);
        new Thread(timeServer, "AIO-AsyncTimeServerHandler-001").start();
    }
}

import java.io.IOException;
import java.io.InterruptedIOException;
import java.net.InetSocketAddress;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.util.concurrent.CountDownLatch;

/**
 * Created by ning on 2017/6/10.
 */
public class AsyncTimeServerHandler implements Runnable{
    private int port;
    //CountDownLatch对象的作用是在完成一组正在执行的操作之前，允许当前的线程一直阻塞
    CountDownLatch latch;
    //异步服务器通道
    AsynchronousServerSocketChannel asynchronousServerSocketChannel;

    public AsyncTimeServerHandler(int port){
        this.port = port;
        try{
            //打开通道，绑定监听端口
            asynchronousServerSocketChannel = AsynchronousServerSocketChannel.open();
            asynchronousServerSocketChannel.bind(new InetSocketAddress(port));
        }catch (IOException e){

        }
    }

    /**
     *
     */
    @Override
    public void run(){
        latch = new CountDownLatch(1);
        doAccept();
        try{
            latch.await();
        }catch (InterruptedException e){
            e.printStackTrace();
        }
    }

    /**
     * 用于接受客户端的链接，由于是异步操作，我们可以传递一个CompletionHandler<AsynchronusSocketChannel,? super A>
     * 类型的handler实例接受accept操作成功的通知消息
     */
    public void doAccept(){
        asynchronousServerSocketChannel.accept(this, new AcceptCompletionHandler());
    }
}

import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;

/**
 * Created by ning on 2017/6/10.
 */
public class AcceptCompletionHandler implements CompletionHandler<AsynchronousSocketChannel, AsyncTimeServerHandler> {
    /**
     * 既然已经接受客户端成功了，为什么还要再次调用accept方法呢?原因是这样的：调用AsynchronousServerSocketChannel的
     * accept方法后，如果有新的客户端连接接入，系统将回调我们传入的CompletionHandler实例的completed方法，表示新的客户端
     * 已经接入成功了。因为一个AsynchronousServerSocket Channel可以接收成千上万个客户端，所以需要继续调用它的accept方法，
     * 接收其他的客户端连接，最终形成一个循环。每当接受一个客户端连接成功之后，再异步接收新的客户端连接。
     */
    @Override
    public void completed(AsynchronousSocketChannel result, AsyncTimeServerHandler attachment){
        attachment.asynchronousServerSocketChannel.accept(attachment, this);
        //链路建立成功之后，服务端需要接收客户端的请求信息，这里创建新的ByteBuffer, 预备分配1MB的缓冲区
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        /**
         * 进行异步读操作，read参数解析
         * ByteBuffer dst: 接收缓冲区，用于从异步Channel中读取数据包
         * A attachment: 异步Channel携带的附件，通知回调的时候作为入参使用
         * CompletionHandler<Integer,? super A>: 接收通知回调的业务Handle
         */
        result.read(buffer, buffer, new ReadCompletionHandler(result));
    }

    @Override
    public void failed(Throwable exc, AsyncTimeServerHandler attachment){
        exc.printStackTrace();
        attachment.latch.countDown();
    }
}

AIO创建的TimeClient

/**
 * Created by ning on 2017/6/10.
 */
public class TimeClient {
    public static void main(String[] args){
        Integer port = 8080;
        if(args != null && args.length > 0) {
            try {
                port = Integer.valueOf(args[0]);
            } catch (NumberFormatException e) {
                //采用默认值
            }
        }
        new Thread(new AsyncTimeClientHandler("127.0.0.1", port), "AIO-AsyncTimeClientHandler-001").start();
    }
}

import java.io.IOException;
import java.io.InterruptedIOException;
import java.io.UnsupportedEncodingException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.util.concurrent.CountDownLatch;

/**
 * Created by ning on 2017/6/10.
 */
public class AsyncTimeClientHandler implements CompletionHandler<Void, AsyncTimeClientHandler>, Runnable {
    private AsynchronousSocketChannel client;
    private String host;
    private int port;
    private CountDownLatch latch;

    public AsyncTimeClientHandler(String host, int port){
        this.host = host;
        this.port = port;
        try{
            client = AsynchronousSocketChannel.open();
        }catch (IOException e){
            e.printStackTrace();
        }
    }

    @Override
    public void run(){
        latch = new CountDownLatch(1);
        client.connect(new InetSocketAddress(host, port), this, this);
        try {
            latch.await();
        }catch (InterruptedException e1){
            e1.printStackTrace();
        }
        try {
            client.close();
        }catch (IOException e){
            e.printStackTrace();
        }
    }

    @Override
    public void completed(Void result, AsyncTimeClientHandler attachment){
        byte[] req = "QUERY TIME ORDER".getBytes();
        final ByteBuffer writeBuffer = ByteBuffer.allocate(req.length);
        writeBuffer.put(req);
        writeBuffer.flip();
        client.write(writeBuffer, writeBuffer, new CompletionHandler<Integer, ByteBuffer>() {
            public void completed(Integer result, ByteBuffer buffer) {
                if(buffer.hasRemaining()){
                    client.write(buffer, buffer, this);
                }else{
                    ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                    client.read(readBuffer, readBuffer, new CompletionHandler<Integer, ByteBuffer>() {
                        public void completed(Integer result, ByteBuffer buffer) {
                            buffer.flip();
                            byte[] bytes = new byte[buffer.remaining()];
                            buffer.get(bytes);
                            String body;
                            try{
                                body = new String(bytes, "UTF-8");
                                System.out.println("Now is :" + body);
                                latch.countDown();
                            }catch (UnsupportedEncodingException e){
                                e.printStackTrace();
                            }
                        }

                        public void failed(Throwable exc, ByteBuffer attachment) {
                            try {
                                client.close();
                            }catch (IOException e){

                            }
                        }
                    });
                }
            }

            public void failed(Throwable exc, ByteBuffer attachment) {
                try{
                    client.close();
                    latch.countDown();
                }catch (IOException e){

                }
            }
        });

    }

    @Override
    public void failed(Throwable exc, AsyncTimeClientHandler attachment){
        exc.printStackTrace();
        try{
            client.close();
            latch.countDown();
        }catch (IOException e){
            e.printStackTrace();
        }
    }
}

结果:

下面继续看下JDK异步回调CompletionHandler的线程执行堆栈：

从”Thread-4”线程堆栈中可以发现，JDK底层通过线程池ThreadPoolExecutor来执行回调通知，异步回调通知类由sun.nio.ch.AsynchronousChannelGroupImpl实现，它经过层层调用，最终回调AsyncTimeClientHandler$1.completed方法，完成回调通知。由此可以得出结论：异步Socket Channel是被动执行对象，我们不需要像NIO编程那样创建一个独立的I/O线程来处理读写操作。对于AsynchronousServerSocketChannel和AsynchronousSocketChannel，它们都由JDK底层的线程池负责回调并驱动读写操作。正因为如此，基于NIO2.0新的异步非阻塞Channel进行编程比NIO编程更为简单。

4种I/O的对比