第11章 - Java NIO

作者：vwFisher

时间：2019-09-04

GitHub代码：https://github.com/vwFisher/JavaBasicGuide

目录

• 1. Java NIO
  • 1.1 传统的BIO编程
  • 1.2 伪异步I/O编程
  • 1.3 NIO编程
    • 1.3.1 Reactor模型
    • 1.3.2 核心类库
      • 1.3.2.1 Buffer - 缓存区
      • 1.3.2.2 Channel - 通道
      • 1.3.2.3 Selector - 多路复用器
      • 1.3.2.4 SelectKey
    • 1.3.3 NIO服务端
    • 1.3.4 NIO客户端
    • 1.3.5 NIO的数据访问方式
  • 1.4 AIO编程
    • 1.4.1 AIO创建的TimerServer源码分析
    • 1.4.2 AIO创建的TimerClient源码分析
  • 1.5 四种I/O的对比

1. Java NIO

在旧的Java版本中所有Socket通信都采用了同步阻塞模式(BIO)，这种一请求一应答模型简化了上层的应用开发。但是在性能和可靠性方面却存在着巨大的瓶颈。JDK1.4提供了新的NIO类库，支持非阻塞IO

NIO提供很多异步API，主要类和接口如下：

1. 进行异步I/O操作的缓冲区ByteBuffer等;
2. 进行异步I/O操作的管道Pipe;
3. 进行各种I/O操作(异步或者同步)的Channel, 包括ServerSocketChannel和SocketChannel;
4. 多种字符集的编码能力和解码能力;
5. 实现非阻塞I/O操作的多路复用器Selector
6. 基于主流的Perl实现的正则表达式类库
7. 文件通道的FileChannel

NIO是基于时间驱动思想来实现的，它采用Reactor(多路复用)模式实现，主要用来解决BIO(同步阻塞IO)模型中一个服务端无法同时并发处理大量客户端连接的问题。

基于Selector进行轮询, 当Socket有数据可读、可写、连接完成、新的TCP请求接入事件时, 操作系统内核会出发Selector返回准备就绪的SelectionKey的集合, 通过SelectableChannel进行读写操作. 由于JDK的Selector底层基于epoll实现, 因此不受2048连接数的限制, 理论上可以同时处理操作系统最大文件句柄个数的连接。SelectableChannel的读写操作都是异步非阻塞的, 当由于数据没有就绪导致读半包时, 立即返回, 不会同步阻塞等待数据就绪, 当TCP缓冲区数据就绪之后, 会触发Selector的读事件, 驱动下一次读操作. 因此, 一个Reactor线程就可以同时处理N个客户端的连接, 这就解决了之前BIO的一连接一线程的弊端, 使Java服务端的并发读写能力得到极大的提升

在Java API中提供了两套NIO，一套是针对标准输入输出NIO，另一套就是网络编程NIO

流与块的比较

NIO和IO最大的区别是数据打包和传输方式。IO是以流的方式处理数据，而NIO是以块的方式处理数据。

面向流的IO一次一个字节的处理数据，一个输入流产生一个字节，一个输出流就消费一个字节。为流式数据创建过滤器就变得非常容易，链接几个过滤器，对数据进行处理非常方便而简单，但是面向流的IO通常处理的很慢。

面向块的IO系统以块的形式处理数据。每一个操作都在一步中产生或消费一个数据块。按块要比按流快的多，但面向块的IO缺少了面向流IO所具有的优雅性和简单性。

1.1 传统的BIO编程

网络编程的基本模型是Client/Server模型, 也就是两个进程之间进行相互通信, 其中服务端提供位置信息(绑定的IP地址和监听端口), 客户端通过连接操作向服务端监听的地址发起连接请求, 通过三次握手建立连接, 如果连接建立成功,双发就可以通过网络套接字(Socket)进行通信

在基于传统同步阻塞模型开发中, ServerSocket负责绑定IP地址, 启动监听端口; Socket负责发起连接操作. 连接成功之后, 双方通过输入和输出流进行同步阻塞式通信

在JDK1.4推出JAVA NIO之前, 基于Java的所有Socket通信都采用了同步阻塞模式(BIO), 这种, 一请求一应答的通信模型简化了上层的应用开发, 但是在可靠性和性能方面存在巨大的瓶颈

下图所示的通信模型图来熟悉BIO的服务端通信模块; 采用BIO通信模型的服务端, 通常由一个独立的Acceptor线程负责监听客户端的连接, 接收到客户端连接之后为客户端连接创建一个新的线程来处理请求信息, 处理完成之后, 返回应答消息给客户端, 线程销毁, 这就是典型的一个请求、一个应答的通信模型.

BIO通信模型图.png

BIO主要问题：每当有一个新的客户端的请求接入时, 服务端必须创建一个新的线程处理新键入的客户端链路, 一个线程只有处理一个客户端连接。成1:1的线形正比

由于线程是Java虚拟机非常宝贵的系统资源, 当线程数膨胀之后, 系统的性能急剧下降, 随着并发量的继续增加, 可能会发生句柄溢出、线程堆栈溢出等问题, 并导致服务器最终宕机或者僵死, 不能对外提供服务.

为了改进一线程一连接的模型, 后来又演进出一种通过线程池或者消息队列实现1个或者多个线程处理N个客户端的模型, 由于它的底层通信机制依然使用同步阻塞I/O, 所以被称为"伪异步"

1.2 伪异步I/O编程

伪异步I/O模型，为了解决同步阻塞I/O面临的一个链路需要线程处理的问题，通过一个线程池来处理多个客户端的请求接入, 形成客户端个数M : 线程池最大线程数N的比例关系, 其中M可以远远大于N. 通过线程池可以灵活地调配线程资源, 设置线程的最大值, 防止由于海量并发接入导致线程耗尽。

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

伪异步服务端框架.png

伪异步通信框架能够缓解BIO面临的问题, 但是无法从根本上解决问题, 由于IO到读写操作会被阻塞, 当并发量增加或者网络IO时延增大之后, 线程的执行时间会被拉长, 它导致缓存在任务队列中的任务不断堆积, 最终导致内存溢出或者拒绝新任务的执行.

由于网络的时延、客户端的执行速度和服务器的处理能力不同, 导致网络IO的执行时间不可控, 如果IO读写被阻塞, 阻塞时间往往也是不可控的(或者超时), 它会导致IO线程的不可预期性阻塞, 降低系统的处理能力和网络吞吐量.在大规模高并发、高性能的服务器端, 使用Java的同步IO来构建服务器是无法满足性能、可扩展性和可靠性要求的

1.3 NIO编程

新的输入/输出(NIO)库是在JDK1.4中引入的. NIO弥补了原来同步阻塞I/O的不足, 它在标准Java代码中提供了高速的、面向块的I/O. 通过定义包含数据的类, 以及通过以块的形式处理这些数据, NIO不用使用本机代码就可以利用低级优化, 这是原来I/O包所无法做到的.

与Socket类和ServerSocket类相对应, NIO也提供了SocketChannel和ServerSocketChannel两种不同的Socket(套接字)通道实现. 这两种新增的通道都支持阻塞和非阻塞两种模式.

1. 阻塞模式: 使用非常简单, 但是性能和可靠性都不好. 适合低负载、低并发的应用程序可以选择同步阻塞I/O, 以降低编程复杂度
2. 非阻塞模式: 适合高负载、高并发的网络应用, 需要使用NIO的非阻塞模式进行开发

基于Reactor模式实现的多路非阻塞高性能的网络IO。

1.3.1 Reactor模型

Reactor模式，由事件驱动，在应用中，将一个请求的能够分离并且调度给应用程序。

简单的说：就是对于一个请求的多个事件(如连接、读写等)，经过这种模式的处理，能够区分出来，并且分别交给对应的处理模块处理。来看下一个简图

Reactor通信模型.png

可以看到Reactor模式中组件有acceptor、dispatcher和handler

• acceptor：注册了各类事件, 当连接有新的事件过来时, 其会将事件交给dispatcher进行分发;
• dispatcher：绑定了事件和对应处理程序handler的映射关系, 当接到新事件时会分发到对应handler;
• handler：负责处理对应事件, 这块就是我们的业务层了

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

1.3.2 核心类库

NIO有3个核心对象：Selector、Buffer、Channel，几乎每一个IO操作中都会用到它们

1. Buffer(缓存区)：缓存区，数据容器。发送给Channel，和Channel中读取书读都要通过Buffer
2. Channel(通道)：通道，任何来源和目的数据都必须通过Channel对象
3. Selector(多路复用器)：负责轮询Channel的状态

1.3.2.1 Buffer - 缓存区

我们首先介绍缓冲区(Buffer)的概念, Buffer是一个对象, 它包含了一些要写入或者要读出的数据.

在面向流的IO中，可以将数据直接写入或者将数据直接读到Stream对象中

在NIO中，加入了Buffer对象，所有数据都是用缓存区处理的。读写都是通过缓冲区进行操作的

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

Buffer
|-- ByteBuffer 字节缓冲区，操作byte数组
    |-- HeadpByteBuffer
    |-- MappedByteBuffer
|-- CharBuffer 字符缓冲区
|-- ShortBuffer 短整型缓冲区
|-- IntBuffer 整型缓冲区
|-- LongBuffer 长整型缓冲区
|-- FloatBuffer 浮点型缓冲区
|-- DoubleBuffer 双精度浮点型缓冲区

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

下表提供Buffer属性和方法

属性

实例化方法

ByteBuffer allocate(int capacity)   
    从JCM堆空间中分配一个容量大小为capacity的byte数组作为缓冲区的byte数据存储器
  
ByteBuffer allocateDirect(int capacity)
    从操作系统来创建内存块用作缓冲区，与当前操作系统能够更好的耦合，能高I/O操作速度。但是分配直接缓冲区的系统开销很大, 因此只有在缓冲区较大并长期存在, 或者需要经常重用时, 才使用这种缓冲区

ByteBuffer wrap(byte[] array)
    这个缓冲区的数据会存放在byte数组中, bytes数组或buff缓冲区任何一方中数据的改动都会影响另一方. 其实ByteBuffer底层本来就有一个bytes数组负责来保存buffer缓冲区中的数据, 通过allocate方法系统会帮你构造一个byte数组

ByteBuffer wrap(byte[] array, int offset, int length)
    可以指定偏移量和长度, 这个offset也就是包装后byteBuffer的position, 而length就是limit-position的大小, 从而我们可以得到limit的位置为length+position(offset)

put()有对应基础类型的put方法(如putInt、putLong)，get同理。在Buffer中占用对应基础类型的位数。

通常使用 Buffer 读写数据一般遵循以下四个步骤:

1. 写入数据到 Buffer；
2. 调用 flip() 方法；
3. 从 Buffer 中读取数据；
4. 调用 clear() 方法或者 compact() 方法。

当向 Buffer 写入数据时, Buffer 会记录下写了多少数据. 一旦要读取数据, 需要通过 flip() 方法将 Buffer 从写模式切换到读模式. 在读模式下, 可以读取之前写入到 Buffer 的所有数据.

一旦读完了所有的数据，就需要清空缓冲区，让它可以再次被写入。有两种方式能清空缓冲区：调用 clear() 或 compact() 方法。clear() 方法会清空整个缓冲区。compact() 方法只会清除已经读过的数据。任何未读的数据都被移到缓冲区的起始处，新写入的数据将放到缓冲区未读数据的后面。

DireactByteBuffer和Non-Direct Buffer(HeapByteBuffer)对比

1.3.2.2 Channel - 通道

Channel是一个通道，它就像自来水管一样，网络数据通过Channel读取和写入Buffer。可以看作是IO的流，但是它和流相比还有一些不同：

1. Channel是双向工的，读写可以二者同时进行，而流是单向的
2. Channel可以进行异步的读写
3. 对Channel的读写必须通过buffer对象

因为Channel是双向的，所以Channel可以比流更好地反映出底层操作系统的真实情况。特别是在Unix模型中，底层操作系统通常都是双向的。

在Java NIO中Channel主要有如下几种类型：

1. FileChannel：从文件读取数据的
2. DatagramChannel：读写UDP网络协议数据
3. SocketChannel：读写TCP网络协议数据
4. ServerSocketChannel：可以监听TCP连接

1.3.2.3 Selector - 多路复用器

多路复用器Selector, 它是Java NIO编程的基础, 熟练地掌握Selector对于NIO编程至关重要. 多路复用器提供选择已经就绪的任务的能力.

在Selector可以注册多个Channel，Selector会监听(不断轮询)询注册在Channel上的事件，如果某个Channel上面发生读或者写事件，这个Channel就处于就绪状态，会被Selector轮询出来，然后通过SelectionKey可以获取就绪Channel的集合，进行后续的I/O事件

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

一旦向Selector注册了一或多个通道, 就可以调用几个重载的select()方法. 这些方法返回你所感兴趣的事件(如连接、接受、读或写)已经准备就绪的那些通道. 换句话说, 如果你对"Read Ready"的通道感兴趣, select()方法会返回读事件已经就绪的那些通道:

1).int select(): 阻塞到至少有一个通道在你注册的事件上就绪
2).int select(long timeout)：select()一样, 除了最长会阻塞timeout毫秒(参数)
3).int selectNow(): 不会阻塞, 不管什么通道就绪都立刻返回, 此方法执行非阻塞的选择操作. 如果自从前一次选择操作后, 没有通道变成可选择的, 则此方法直接返回零.

select()方法返回的int值表示有多少通道已经就绪. 亦即, 自上次调用select()方法后有多少通道变成就绪状态. 如果调用select()方法, 因为有一个通道变成就绪状态, 返回了1, 若再次调用select()方法, 如果另一个通道就绪了, 它会再次返回1. 如果对第一个就绪的channel没有做任何操作, 现在就有两个就绪的通道, 但在每次select()方法调用之间, 只有一个通道处于就绪状态

某个线程调用select()方法后阻塞了, 即使没有通道就绪, 也有办法让其从select()方法返回. 只要让其它线程在第一个线程调用select()方法的那个对象上调用Selector.wakeup()方法即可. 阻塞在select()方法上的线程会立马返回

如果有其它线程调用了wakeup()方法, 但当前没有线程阻塞在select()方法上, 下个调用select()方法的线程会立即"醒来(wake up)"

当用完Selector后调应道掉用close()方法, 它将关闭Selector并且使注册到该Selector上的所有SelectionKey实例无效. 通道本身并不会关闭

1.3.2.4 SelectKey

通过调用Channal的register()来注册到Selector

public final SelectionKey register(Selector sel, int ops) throws ClosedChannelException

注册的Channel必须设置成异步模式才可以, 否则异步IO就无法工作, 这就意味着我们不能把一个FileChannel注册到Selector, 因为FileChannel没有异步模式, 但是网络编程中的SocketChannel是可以的.

register的第二个参数：是一个int类型的ops，代表注册的Selector对Channel中的哪些事件感兴趣, 事件类型有四种，是在SelectionKey定义的常量

SelectionKey定义事件常量如下

在SelectionKey中可以获取Selector和注册的Channel

Channel  channel  = selectionKey.channel();
Selector selector = selectionKey.selector(); 

1.3.3 NIO服务端

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

NIO服务端时序图.png

(basic.nio.nio.TimeServer)

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.text.SimpleDateFormat;
import java.util.Date;
import java.util.Iterator;
import java.util.Set;

public class TimeServer {
    private static int port = 8080;
    public static void main(String[] args) throws IOException {
        // 创建一个多路复用类, 独立的线程, 负责轮询多路复用器Selector, 可以处理多个客户端的并发接入
        ReactorThread reactorThread = new ReactorThread(port);
        new Thread(reactorThread, "NIO-ReactorThread").start();
    }
    
    /** Reactor 处理器 */
    static class ReactorThread implements Runnable {
        private static String QUERY_TIME_ORDER_STR = "QUERY TIME ORDER";
        private static String BAD_ORDER_STR = "BAD ORDER";
        private Selector selector;
        private ServerSocketChannel serverChannel;
        private volatile boolean stop;

        /** 初始化多路复用器, 资源初始化。创建多路复用器Selector, ServerSocketChannel, 对Channel和TCP参数进行配置 */ 
        public ReactorThread(int port) {
            try {
                // 1).打开ServerSocketChannel, 用于监听客户端的连接, 它是所有客户端连接的父管道
                serverChannel = ServerSocketChannel.open();
                // 2).设置为异步非阻塞模式, 绑定指定端口, 它的backlog设为1024
                serverChannel.configureBlocking(false);
                serverChannel.socket().bind(new InetSocketAddress(port), 1024);
                // 3).创建Reactor线程, 创建多路复用器并启动线程
                selector = Selector.open();
                // 4).将ServerSocketChannel注册到Selector, 监听SelectionKey.OP_ACCEPT操作位
                serverChannel.register(selector, SelectionKey.OP_ACCEPT);
                System.out.println("--> The TimeServer is start in port:" + port);
            } catch (IOException e) {
                e.printStackTrace();
                System.exit(1);
            }
        }
        
        public void stop() {
            this.stop = true;
        }
        
        @Override
        public void run() {
            // 多路复用器在线程run方法的无限循环体内轮询准备就绪的Key
            while (!stop) {
                try {
                    /**
                     * 休眠时间为1s, 无论是否有读写等事件发生, selector每隔1s被唤醒一次, selector也提供了无参的select方法;
                     * 当有处于就绪状态的Channel时, selector将返回改Channel的SelectionKey集合, 
                     * 通过对就绪状态的Channel集合进行迭代, 可以进行网络的异步读写操作
                     */
                    selector.select(1000);
                    Set<SelectionKey> selectedKeys = selector.selectedKeys();
                    Iterator<SelectionKey> it = selectedKeys.iterator();
                    SelectionKey key = null;
                    while (it.hasNext()) {
                        // 5).轮询SelectionKey，处理IO事件
                        key = it.next();
                        it.remove();
                        try {
                            handleSelectionKey(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();
                }
            }
        }
        
        /** 处理SelectionKey */
        private void handleSelectionKey(SelectionKey key) throws IOException {
            // 处理新接入的请求信息, 根据SelectionKey的操作位进行判断即可获知网络事件的类型
            if (key.isAcceptable()) {
                // Accept the new connection
                ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
                // 6).通过ServerSocketChannel的accept接收客户端的连接请求并创建SocketChannel实例(完成TCP三次握手, 建立物理链路)
                SocketChannel sc = ssc.accept();
                // 7).需要将新创建的SocketChannel设置为异步非阻塞, 同时也可以对其TCP参数进行设置
                sc.configureBlocking(false);
                // 8).注册READ事件，Add the new connection to the selector
                sc.register(selector, SelectionKey.OP_READ);
            }
            if (key.isReadable()) {
                // Read the data
                SocketChannel sc = (SocketChannel) key.channel();
                // 开辟1MB的缓冲区
                ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                // 9).异步调用read方法读取请求码流到缓冲区
                int readBytes = sc.read(readBuffer);
                if (readBytes > 0) {
                    // 10).读到了字节, 对字节进行编解码
                    readBuffer.flip();
                    byte[] bytes = new byte[readBuffer.remaining()];
                    readBuffer.get(bytes);
                    String body = new String(bytes, "UTF-8");
                    System.out.println("--> The TimeServer receive order:" + body);
                    String currentTime = BAD_ORDER_STR;
                    if (QUERY_TIME_ORDER_STR.equalsIgnoreCase(body)) {
                        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                        currentTime = sdf.format(new Date());
                    }
                    // 11).将消息异步发送给客户端, 示例代码如下
                    doWrite(sc, currentTime);
                } else if (readBytes < 0) {
                    // 对端链路关闭, 需要关闭SocketChannel, 释放资源
                    key.cancel();
                    sc.close();
                } else {
                    ;   // 没有读取到字节, 属于正常场景, 忽略
                }
            }
        }
        
        private void doWrite(SocketChannel channel, String response) throws IOException {
            if (response != null && response.trim().length() > 0) {
                byte[] bytes = response.getBytes();
                // 根据字节数组的容量创建ByteBuffer
                ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
                // 将字节数组赋值到缓冲区中
                writeBuffer.put(bytes);
                // 对缓冲区进行flip操作: java.nio.HeapByteBuffer[pos=0 lim=9 cap=9]
                writeBuffer.flip();
                // 将缓冲区中的字节数组发送出去
                channel.write(writeBuffer);
            }
        }
    }
}

由于这个类相比于传统的Socket编程会稍微复杂一些, 在此展开进行详细分析, 从如下几个关键步骤来讲解多路复用处理类.

    1).构造方法, 在构造方法中进行资源初始化. 创建多路复用器Selector、ServerSocketChannel, 对Cahnnel和TCP参数进行配置.(例如, 将ServerSocketChannel设置为异步非阻塞模式, 它的backlog设为1024). 系统资源初始化成功后, 将ServerSocketChannel注册到Selector, 监听SelectionKey.OP_ACCEPT操作位. 如果资源初始化失败(例如端口被占用), 则退出.
    2).在线程的run方法的while循环体重循环遍历selector, 它的休眠时间为1s. 无论是否有读写等事件发生, selector每隔1s都被唤醒一次. selector也提供了一个无参的select方法: 当有处于就绪状态的Channel时, selector将返回改Channel的SelectionKey集合. 通过对就绪状态的Channel集合进行迭代, 可以记性网络的异步读写操作.
    3).处理新接入的客户端请求消息, 根据Selectionkey的操作位进行判断即可获知网络事件的类型, 通过ServerSocketChannel的accept接收客户端的连接请求并创建SocketChannel实例, 完成上述操作后, 相当于完成了TCP的三次握手, TCP物理链路正式建立. 
[注: 我们需要将新创建的SocketChannel设置为异步非阻塞, 同时也可以对其TCP参数进行设置. 例如TCP接收和发送缓冲区大小等. 但作为入门的例子, 以上例子没有进行额外的参数设置]
    4).用于读取客户端的请求消息. 首先创建一个ByteBuffer. 由于我们事先无法得知客户端发送的码流大小, 作为流程, 我们开辟一个1MB的缓冲区. 然后调用SocketChannel的read方法读取请求码流. 注意, 由于我们将SocketChannel设置为异步非阻塞模式, 因此它的read是非阻塞的. 使用返回值进行判断, 看读取到的字节数, 返回值有以下三种可能的结果.
    {1}.返回值 > 0: 读到了字节, 对字节进行编解码;
    {2}.返回值 = 0: 没有读取到字节, 属于正常场景, 忽略;
    {3}.返回值 = -1: 链路已经关闭, 需要关闭SocketChannel, 释放资源
    当读取到码流之后, 进行解码. 首先对readBuffer进行flip操作, 它的作用是将缓冲区当前的limit设置为position, position设置为0, 用于后续对缓冲区的读取操作. 然后根据缓冲区可读的字节个数创建字节数组, 调用ByteBuffer的get操作将缓冲区可读的字节数组复制到新创建的字节数组中, 最后调用非富川的构造函数创建请求消息体并打印, 如果请求指令是"QUERY TIME ORDER", 则把服务器的当前时间编码后返回给客户端. 下面我们看看异步发送应答消息给客户端的情况.
    5).将应答消息异步发送给客户端. 我们看下关键代码, 首先将字符串编码成字节数组, 根据字节数组的容量创建ByteBuffer, 调用ByteBuffer的put操作字节数组赋值到缓冲区中, 然后对缓冲区进行flip操作, 最后调用SocketChannel的write方法将缓冲区中的字节数组发送出去. 需要指出的时, 由于SocketChannel是异步非阻塞的, 它并不保证一次能够把需要发送的字节数组发送完, 此时会出现"写半包"问题, 我们需要注册写操作, 不断轮询Selector将没有发送完的ByteBuffer发送完毕, 然后可以通过ByteBuffer的hasRemain()方法判断下次是否发送完成

1.3.4 NIO客户端

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

NIO客户端时序图.png

(basic.nio.nio.TimeClient)

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.util.Iterator;
import java.util.Set;

public class TimeClient {
    private static String host = "127.0.0.1";
    private static int port = 8080;
    
    public static void main(String[] args) throws IOException {
        new Thread(new TimeClientHandle(host, port), "TimeClient-001").start();
    }
    
    static class TimeClientHandle implements Runnable {
        private String host;
        private int port;
        private Selector selector;
        private SocketChannel socketChannel;
        private volatile boolean stop;
        
        public TimeClientHandle(String host, int port) {
            this.host = host == null ? "127.0.0.1" : host;
            this.port = port;
            try {
                // 1).打开ServerSocketChannel, 用于监听客户端的连接, 它是所有客户端连接的父管道
                socketChannel = SocketChannel.open();
                // 2).设置为异步非阻塞模式
                socketChannel.configureBlocking(false);
                // 6).创建Reactor线程, 创建多路复用器并启动线程
                selector = Selector.open();
            } catch (IOException e) {
                e.printStackTrace();
                System.exit(1);
            }
        }

        @Override
        public void run() {
            try {
                doConnect();
            } catch (IOException e) {
                e.printStackTrace();
                System.exit(1);
            }
            // 7).多路复用器在线程run方法的无限循环体内轮询准备就绪的Key
            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 {
                            handleSelectionKey(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();
                }
            }
        }
        
        /** 连接，注册OP_READ和OP_CONNECT */
        private void doConnect() throws IOException {
            // 4).如果直接连接成功, 则注册到多路复用器上, 发送请求信息, 读应答
            if (socketChannel.connect(new InetSocketAddress(host, port))) {
                socketChannel.register(selector, SelectionKey.OP_READ);
                doWrite(socketChannel);
            } else {    
                // 向Reactor线程的多路复用器注册OP_CONNECT状态为, 监听服务端的TCP ACK应答
                socketChannel.register(selector, SelectionKey.OP_CONNECT);
            }
        }
        
        private void handleSelectionKey(SelectionKey key) throws IOException {
            if (key.isValid()) {
                // 判断是否连接成功
                SocketChannel sc = (SocketChannel) key.channel();
                // 8).接收connect事件并进行处理
                if (key.isConnectable()) {
                    // 9).判断连接结果, 如果连接成功, 注册读事件到多路复用器
                    if (sc.finishConnect()) {
                        // 10). 注册读事件到多路复用器
                        sc.register(selector, SelectionKey.OP_READ);
                        doWrite(sc);
                    } else {
                        // 连接失败, 进程退出
                        System.exit(1); 
                    }
                }
                if (key.isReadable()) {
                    // 开辟1MB的缓冲区
                    ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                    // 11).异步调用read方法读取请求码流到缓冲区
                    int readBytes = sc.read(readBuffer);
                    if (readBytes > 0) {
                        // 12).读到了字节, 对字节进行编解码
                        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) {
                        // 对端链路关闭, 需要关闭SocketChannel, 释放资源
                        key.cancel();
                        sc.close();
                    } else {
                        ;   // 没有读取到字节, 属于正常场景, 忽略
                    }
                }
            }
        }
        
        private void doWrite(SocketChannel sc) throws IOException {
            byte[] req = "QUERY TIME ORDER".getBytes();
            ByteBuffer writeBuffer = ByteBuffer.allocate(req.length);
            writeBuffer.put(req);
            // 对缓冲区进行flip操作: java.nio.HeapByteBuffer[pos=0 lim=9 cap=9]
            writeBuffer.flip();
            sc.write(writeBuffer);
            if (!writeBuffer.hasRemaining()) {
                System.out.println("Send order to server succeed.");
            }
        }
    }
}

与服务端类似, 接下来我们通过对关键步骤的源码进行分析和解读, 让大家深入了解如何创建NIO客户端以及如何使用NIO的API.

    1).构造函数用于初始化NIO的多路复用器和SocketChannel对象. 需要注意的是, 创建SocketChannel之后, 需要将其设置为异步非阻塞模式. 并可以设置SocketChannel的TCP参数, 例如接收和发送的TCP缓冲区大小
    2).用于发送连接请求, 作为示例, 连接时成功的, 所以不需要做重连操作, 因此将其放到循环之前. 下面我们具体看看doConnect的实现
    对应3).首先对SocketChannel的connect()操作进行判断. 如果连接成功, 则将SocketChannel注册到多路复用器Selector上, 注册SelectionKey.OP_READ; 如果没有直接连接成功, 则说明服务器没有返回TCP握手应答信息, 但这不代表连接失败. 我们需要将SocketChannel注册到多路复用器Selector上, 注册SelectionKey.OP_CONNECT, 当服务端返回TCP syn-ack消息后, Selector就能够轮询到这个SocketChannel处于连接就绪状态
    4).在循环体重轮询多路复用器Selector. 当有就绪的Channel时, 执行5).的handleInput(key)方法. 下面我们对handleInput()方法进行分析
    跳到6).我们首先对SelectionKey进行判断, 看它处于什么状态. 如果是处于连接状态, 说明服务端已经返回ACK应答消息. 这时我们需要对连接结果进行判断, 调用SocketChannel的finishConnection()方法, 如果返回true, 说明客户端连接成功; 如果返回为false或直接抛出IOException, 说明连接失败. 连接成功后将SocketChannel注册到多路复用器上, 注册SelectionKey.OP_READ操作位, 监听网络读操作, 然后发送请求消息给服务端, 下面我们对doWrite(sc)进行分析;
    代码跳到7).我们构造请求消息体, 然后对其编码, 写入到发送缓冲区中, 最后调用SocketChannel的write方法进行发送. 由于发送时异步的, 所以会存在"半包写"问题, 此处不再赘述. 最后通过hasRemaining()方法对发送结果进行判断, 如果缓冲区中的消息全部发送完成, 打印"Send order to server succeed."
    5).返回代码8).我们继续分析客户端是如何读取时间服务区应答消息的, 如果客户端接收到了服务端的应答消息, 则SocketChannel是可读的, 由于无法事先判断应答码流的大小, 我们就预分配1MB的接收缓冲区用于读取应答消息, 调用SocketChannel的read()方法进行异步读取操作. 由于是异步操作,所以必须对读取的结果进行判断, 这部分逻辑已经在2.3.3小节详细介绍过, 此处不再赘述. 如果读到了消息, 则对消息进行解码, 最后打印结果. 执行完成后将stop置为true, 线程退出循环.
    6).线程退出循环后: 我们需要对连接资源进行释放, 以实现"优雅退出". 代码9).用于多路复用器的资源释放, 由于多路复用器上可能注册成千上万的Channel或者pipe, 如果一一对这些资源进行释放显然不合适, 因此, JDK底层会自动释放所有跟此多路复用器关联的资源.
    NIO编程的难度确实比同步阻塞的BIO的大很多，而且上面没有考虑"半包读"和"半包写"，对NIO优点总结
    1).客户端发起的连续操作是异步的, 可以通过在多路复用器注册OP_CONNECT等待后续结果, 不需要像之前的客户端那样被同步阻塞
    2).SocketChannel的读写操作都是异步的, 如果没有可读写的数据它不会同步等待, 直接返回, 这样I/O通信线程就可以处理其他的链路, 不需要同步等待这个链路可用
    3).线程模型的优化: 由于JDK的Selector在Linux等主流操作系统上通过epoll实现, 它没有连接句柄数的限制(只受限于系统的最大句柄或者单个进程的句柄限制), 这意味着一个Selector线程可以同时处理成千上万个客户端连接, 而且性能不会随着客户端的增加而线性下降. 因此, 它非常适合做高性能, 高负载的网络服务器.
    JDK1.7升级了NIO类库, 升级后的NIO类库被称为NIO2.0. 引人注目的四, Java正式提供了异步文件I/O操作, 同时提供了与UNIX网络编程事件驱动I/O对应的AIO.

1.3.5 NIO的数据访问方式

NIO提供比传统的文件访问方式更好的办法，NIO有两个优化方法

    FileChannel.transferTo
    FileChannel.transferFrom
    FileChannel.map

1).FileChannel.transferXXX

与传统访问方式相比，减少数据从内核到用户空间的复制，数据直接在内核空间中移动，在Linux使用sendfile系统调用。

2).FileChannel.map

FileChannel将文件按照一定大小块映射为内存区域，当程序访问这个内存区域时，将直接操作这个文件数据。

这种方式省去了数据从内核空间复制到用户空间的损耗。适合对大文件的只读性操作，如大文件的MD5校验。当然这种方式适合操作系统的底层I/O实现相关的

    public static void main(String[] args) {
        int bufferSize = 1024;
        String fileName = "test.db";
        long fileLength = new File(fileName).length();
        int bufferCount = 1 + (int) (fileLength / bufferSize);
        MappedByteBuffer[] buffers = new MappedByteBuffer[bufferCount];
        long remaining = fileLength;
        for (int i = 0; i < bufferCount; i++) {
            RandomAccessFile file;
            try {
                file = new RandomAccessFile(fileName, "r");
                buffers[i] = file.getChannel().map(FileChannel.MapMode.READ_ONLY, i * bufferSize, (int) Math.min(remaining, bufferSize));
            } catch (Exception e) {
                e.printStackTrace();
            }
            remaining -= bufferSize;
        }
    }

1.4 AIO编程

NIO2.0 引入了新的异步通道的概念, 并提供了异步文件通信和异步Socket通道的实现. 异步通道提供了以下两种方式获取操作结果

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

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

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

下面通过代码来熟悉NIO2.0 AIO的相关类库

1.4.1 AIO创建的TimerServer源码分析

AIO对应ServerSocketChannel和SocketChannel，使用AsynchronousServerSocketChannel和AsynchronousSocketChannel

(basic.nio.timeserver.aio.TimeServer)

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

public class TimeServer {
    private static int port = 8080;
    public static void main(String[] args) throws IOException {
        // 创建异步的事件服务器处理类, 然后启动线程将AsyncTimeServerHandler拉起
        AsyncTimeServerHandler timeServer = new AsyncTimeServerHandler(port);
        new Thread(timeServer, "AIO-AsyncTimeServerHandler").start();
    }
    
    static class AsyncTimeServerHandler implements Runnable {
        private int port;
        CountDownLatch latch;
        AsynchronousServerSocketChannel asynchronousServerSocketChannel;
        
        public AsyncTimeServerHandler(int port) {
            this.port = port;
            try {
                // 创建一个异步的服务端通道, 绑定监听端口
                asynchronousServerSocketChannel = AsynchronousServerSocketChannel.open();
                asynchronousServerSocketChannel.bind(new InetSocketAddress(port));
                System.out.println("The TimeServer is start in port:" + port);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        
        @Override
        public void run() {
            /**
             * 初始化CountDownLatch对象, 它的作用是, 在完成一组正在执行的操作之前, 允许当前的线程一直阻塞.
             * 在本例程中, 我们让线程在此阻塞, 防止服务端执行完成退出. 实际项目中, 不需要启动独立的线程来处理AsynchronousServerSocketChannel
             */
            latch = new CountDownLatch(1);
            // 接收客户端的连接, 由于是异步操作, 我们可以传递一个CompletionHandler<AsynchronousSocketChannel, ? super A>类型的handler实例接收accept操作成功的通知消息
            doAccept();
            try {
                latch.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        
        public void doAccept() {
            asynchronousServerSocketChannel.accept(this, new AcceptCompletionHandler());
        }
    }
    
    /** Accept 完成处理器 */
    static class AcceptCompletionHandler implements CompletionHandler<AsynchronousSocketChannel, AsyncTimeServerHandler> {
        /**
         * 会发现外层调用accept(), 此处也调用accept(), 为什么呢?
         *  系统将回调我们传入的CompletionHandler实例的completed方法, 表示新的客户端已经接入成功. 因为一个AsynchronousSocketChannel可以接收成千上万个客户端, 
         * 所以需要继续调用它的accept(), 接收其他的客户端连接,最终形成一个循环. 每当接收一个客户端连续成功之后, 再一步接收新的客户端连接
         */
        @Override
        public void completed(AsynchronousSocketChannel result, AsyncTimeServerHandler attachment) {
            attachment.asynchronousServerSocketChannel.accept(attachment, this);
            // 创建新的ByteBuffer, 预分配1MB的缓冲区
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            /**
             * 调用read进行异步读操作
             * 参数1(ByteBuffer dst): 接收缓冲区, 用于从异步Channel中读取数据包
             * 参数2(A attachment): 异步Channel携带的附件, 通知回调的时候作为入参使用
             * 参数3(CompletionHadnler<Integer, ? super A>): 接收通知回调的业务Handler, 在本例程中为ReadCompletionHandler
             */
            result.read(buffer, buffer, new ReadCompletionHandler(result));
        }

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

    /** Read 完成处理器 */
    static class ReadCompletionHandler implements CompletionHandler<Integer, ByteBuffer> {
        private AsynchronousSocketChannel channel;
        /**
         * 将AsynchronousSocketChannel通过参数传递到ReadCompletionHandler中, 当成员变量使用
         * 作用: 读取半包消息和发送应答
         */
        public ReadCompletionHandler(AsynchronousSocketChannel channel) {
            if (this.channel == null) {
                this.channel = channel;
            }
        }

        @Override
        public void completed(Integer result, ByteBuffer attachment) {
            // 进行flip()操作, 为后续从缓冲区读取数据做准备
            attachment.flip();
            // 根据缓冲区的可读字节创建byte数组
            byte[] body = new byte[attachment.remaining()];
            attachment.get(body);
            try {
                // 通过new String创建请求信息, 读请求消息进行判断
                String req = new String(body, "UTF-8");
                System.out.println("The TimeServer receive order:" + req);
                String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(req) 
                        ? new java.util.Date(System.currentTimeMillis()).toString()
                        : "BAD ORDER";
                doWrite(currentTime);
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
            }
        }
        
        private void doWrite(String currentTime) {
            // 合法性校验, 如果合法, 调用字符串的解码方法将应答信息编码成字节数组
            if (currentTime != null && currentTime.trim().length() > 0) {
                byte[] bytes = currentTime.getBytes();
                ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
                writeBuffer.put(bytes);
                writeBuffer.flip();
                channel.write(writeBuffer, writeBuffer, new CompletionHandler<Integer, ByteBuffer>() {

                    @Override
                    public void completed(Integer result, ByteBuffer buffer) {
                        // 如果没有发送完成, 继续发送
                        if (buffer.hasRemaining()) {
                            channel.write(buffer, buffer, this);
                        }
                    }

                    @Override
                    public void failed(Throwable exc, ByteBuffer attachment) {
                        try {
                            channel.close();
                        } catch (IOException e) {
                            // ingore on close
                        }
                    }
                });
            }
        }

        @Override
        public void failed(Throwable exc, ByteBuffer attachment) {
            try {
                channel.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

首先看构造方法, 我们将AsynchronousSocketChannel通过参数传递到ReadCompletionHanler中, 当做成员变量来使用, 主要用于读取半包消息和发送应答. 本例程不对半包读写进行具体说明, 对此感兴趣的读者可以关注后续章节对Neey半包处理的专题介绍.

1).读取到消息后的处理. 首先对attachment进行flip操作, 为后续从缓冲区读取数据做准备. 根据缓冲区的可读字节数创建byte数组,然后通过new String方法创建请求消息, 对请求消息进行判断, 如果是"QUERY TIME ORDER"则获取当前系统服务器的时间, 调用doWrite方法发送给客户端

doWrite方法实现(2).行), 首先对当前时间进行合法性校验, 如果合法, 调用字符串的解码方法将应答消息编码成字节数组, 然后将它赋值到发送缓冲区writeBuffer中, 最后调用AsynchronousSocketChannel的异步write方法. 正如前面介绍的异步read方法一样, 它也有三个与read方法相同的参数. 在本例程中我们直接实现write方法的异步回调接口CompletionHandler. 代码跳到3).行, 对发送的writeBuffer进行判断, 如果还有剩余的字节可写, 说明没有完成发送, 需要继续发送, 直到发送成功.

最后, 我们关注下failed方法, 它的实现很简单, 就是当发生异常时候, 对异常Throwable进行判断: 如果是I/O异常, 就关闭链路, 释放资源; 如果是其他异常, 按照业务自己的逻辑进行处理. 本例程作为简单的demo, 没有对异常进行分类判断, 只要发生了读写异常, 就关闭链路, 释放资源

异步非阻塞I/O版本的时间服务器服务端已经介绍完毕, 下面看看客户端的实现

1.4.2 AIO创建的TimerClient源码分析

(basic.nio.timeserver.aio.TimeClient)

import java.io.IOException;
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;

public class TimeClient {
    private static String host = "127.0.0.1";
    private static int port = 8080;
    
    public static void main(String[] args) throws IOException {
        // 通过一个独立的I/O线程创建异步时间服务器客户端Handler
        new Thread(new AsyncTimeClientHandle(host, port), "AIO-AsyncThread").start();
    }
    
    static class AsyncTimeClientHandle implements CompletionHandler<Void, AsyncTimeClientHandle>, Runnable {
        private AsynchronousSocketChannel client;
        private String host;
        private int port;
        private CountDownLatch latch;
        
        public AsyncTimeClientHandle(String host, int port) {
            this.host = host;
            this.port = port;
            try {
                // 创建一个新的AsynchronousSocketChannel对象
                client = AsynchronousSocketChannel.open();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        
        @Override
        public void run() {
            // 创建CountDownLatch进行等待, 防止异步操作没有执行完成线程就退出
            latch = new CountDownLatch(1);
            // 通过connect方法发起异步操作
            client.connect(new InetSocketAddress(host, port), this, this);
            try {
                latch.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            try {
                client.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        @Override
        public void completed(Void result, AsyncTimeClientHandle attachment) {
            // 创建请求消息体, 对其进行编码, 然后复制到发送缓冲区
            byte[] req = "QUERY TIME ORDER".getBytes();
            ByteBuffer writeBuffer = ByteBuffer.allocate(req.length);
            writeBuffer.put(req);
            writeBuffer.flip();
            client.write(writeBuffer, writeBuffer, new CompletionHandler<Integer, ByteBuffer>() {
                @Override
                public void completed(Integer result, ByteBuffer buffer) {
                    // 如果发送缓冲区中仍有尚未发送的字节, 将继续异步发送, 如果已经发送完成, 则执行异步读取操作
                    if (buffer.hasRemaining()) {
                        client.write(buffer, buffer, this);
                    } else {
                        ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                        // read方法异步读取服务端的响应消息
                        client.read(readBuffer, readBuffer, new CompletionHandler<Integer, ByteBuffer>() {
                            @Override
                            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();
                                }
                            }
                            @Override
                            public void failed(Throwable exc, ByteBuffer attachment) {
                                try {
                                    client.close();
                                    latch.countDown();
                                } catch (IOException e) {
                                    // ingore on close
                                }
                            }
                        });
                    }
                }

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

        @Override
        public void failed(Throwable exc, AsyncTimeClientHandle attachment) {
            // 当读取发生异常时, 关闭链路, 同时调用CountDownLatch的countDown方法让AsyncTimeClientHandler线程执行完毕, 客户端退出执行
            try {
                client.close();
                latch.countDown();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

由于AsyncTimeClientHandler中大量使用了内部匿名类, 所有代码看起来稍微有些复杂
1).构造函数, 首先通过AsynchronousSocketChannel的open方法创建一个新的AsynchronousSocketChannel对象. 然后调到2).行, 创建CountDownLatch进行等待, 防止异步操作没有执行完成线程就退出, 第3).行通过connect方法发起异步操作, 它有两个参数, 分别是
1).A attachment: AsynchronousSocketChannel的附件, 用于回调通知时作为入参被传递, 调用者可以自定义;
2).CompletionHadnler<Integer, ? super A> handler: 异步操作回调通知接口, 由调用者实现
在本例程中, 这两个参数都使用AsyncTimeClientHandler类本身, 因为它实现了CompletionHandler接口.
接下来我们看异步连接成功之后的方法回调—completed方法, 代码4).行, 我们创建请求消息体, 对其进行编码, 然后复制到发送缓冲区writeBuffer中, 调用AsynchronousSocketChannel的write方法进行异步写. 与服务端类似, 我们可以实现CompletionHandler<Integer, ByteBuffer>接口用于写操作完成后的回调, 代码5).段, 如果发送缓冲区中仍有尚未发送的字节, 将继续异步发送, 如果已经发送完成, 则执行异步读取操作
6).客户端异步读取时间服务器服务端应答消息的处理逻辑, 代码6).行调用AsynchronousSocketChannel的read方法异步读取服务端的响应消息, 由于read操作是异步的, 所以我们通过内部匿名类实现CompletionHandler<Integer, ByteBuffer>接口, 当读取完成被JDK回调时, 构造应答消息. 代码7).段从CompletionHandler的ByteBuffer中读取应答消息并打印结果
8).当读取发生异常时, 关闭链路, 同时调用CountDownLatch的countDown方法让AsyncTimeClientHandler线程执行完毕, 客户端退出执行
JDK底层通过线程池ThreadPoolExecutor来执行回调通知, 异步回调通知类由sun.nio.ch.AsynchronousChannelGroupImpl实现, 它经过层层调用, 最终回调com.phei.netty.aio.AsyncTimeClientHandler$1.completed方法, 完成回调通知. 由此我们也可以得出结论: 异步SocketChannel是被动执行对象, 我们不要想NIO变成那样创建一个独立的I/O线程来处理读写操作. 对于AsynchronousServerSocketChannel和AsynchronousSocketChannel, 他们都由JDK底层的线程池负责回调并驱动读写操作. 正因为如此, 基于NIO2.0新的异步非阻塞Channel进行变成比NIO变成更为简单

1.5 四种I/O的对比

不同的I/O模型由于线程模型、API等差别很大, 所以用法的差异也非常大. 由于之前的几个小节已经集中对这几种I/O的API和用法进行了说明, 本小节会重点对他们进行功能对比. 如表

几种I/O模型的功能和特性对比