Java高级特性系列（四）：好好说说IO其二

【引言】在前一篇关于IO的专题写着写着，发现篇幅越来越大，还没写到具体的BIO、NIO、AIO，文章就已经很长很长了；索性就把这一个专题拆成两部分来写吧，此篇作为IO的第二篇，在前一节建立的理论基础上来总结一下NIO具体的应用类型和特性；此篇重点关注NIO的特性。

NIO的起源和概念

历史起源

  BIO和NIO的分界点，就是JDK1.4。
  在1.4之前，大家都是使用普通的Java IO就是阻塞I/O模式，一个线程只能处理一个流的I/O事件。如果想要同时处理多个流，要么多进程(fork)，要么多线程(pthread_create)，很不幸这两种方法效率都不高。
  在1.4之后推出的同步非阻塞的I/O模型，也是I/O多路复用的基础，已经被越来越多地应用到大型应用服务器，成为解决高并发与大量连接、I/O处理问题的有效方式。

核心数据类型

• Buffer：包含数据且用于读写的线形表结构。其中还提供了一个特殊类用于内存映射文件的I/O操作。
• Charset：它提供Unicode字符串影射到字节序列以及逆映射的操作。
• Channels：包含socket，file和pipe三种管道，都是全双工的通道。
• Selector：多个异步I/O操作集中到一个或多个线程中（可以被看成是Unix中select()函数的面向对象版本）。
  

NIO和IO的区别

• IO：面向流、阻塞IO
• NIO：面向缓冲、非阻塞IO、有选择器特性

面向流与面向缓冲

  Java NIO和IO之间第一个最大的区别是，IO是面向流的，NIO是面向缓冲区的。 Java IO面向流意味着每次从流中读一个或多个字节，直至读取所有字节，它们没有被缓存在任何地方。此外，它不能前后移动流中的数据。如果需要前后移动从流中读取的数据，需要先将它缓存到一个缓冲区。 Java NIO的缓冲导向方法略有不同。数据读取到一个它稍后处理的缓冲区，需要时可在缓冲区中前后移动。这就增加了处理过程中的灵活性。但是，还需要检查是否该缓冲区中包含所有您需要处理的数据。而且，需确保当更多的数据读入缓冲区时，不要覆盖缓冲区里尚未处理的数据。

阻塞与非阻塞IO

  Java IO的各种流是阻塞的。这意味着，当一个线程调用read() 或 write()时，该线程被阻塞，直到有一些数据被读取，或数据完全写入。该线程在此期间不能再干任何事情了。 Java NIO的非阻塞模式，使一个线程从某通道发送请求读取数据，但是它仅能得到目前可用的数据，如果目前没有数据可用时，就什么都不会获取。而不是保持线程阻塞，所以直至数据变的可以读取之前，该线程可以继续做其他的事情。 非阻塞写也是如此。一个线程请求写入一些数据到某通道，但不需要等待它完全写入，这个线程同时可以去做别的事情。 线程通常将非阻塞IO的空闲时间用于在其它通道上执行IO操作，所以一个单独的线程现在可以管理多个输入和输出通道（channel）。

选择器（Selectors）

  Java NIO的选择器允许一个单独的线程来监视多个输入通道，你可以注册多个通道使用一个选择器，然后使用一个单独的线程来“选择”通道：这些通道里已经有可以处理的输入，或者选择已准备写入的通道。这种选择机制，使得一个单独的线程很容易来管理多个通道。

Channel

Channel的定义

  A channel represents an open connection to an entity such as a hardware device, a file, a network socket, or a program component that is capable of performing one or more distinct I/O operations, for example reading or writing.
  NIO把它支持的I/O对象抽象为Channel，Channel又称“通道”，类似于原I/O中的流（Stream），但有所区别：
1、流是单向的，通道是双向的，可读可写。
2、流读写是阻塞的，通道可以异步读写。
3、流中的数据可以选择性的先读到缓存中，通道的数据总是要先读到一个缓存中，或从缓存中写入
  Buffer实际就是缓存，不管是流还是通道，都需要跟Buffer打交道的。Channel和Buffer是紧密联系在一起的，要么写进去，要么读出来：

Channel的实现

[1] Channel和Buffer有好几种类型。下面是JAVA NIO中的一些主要Channel的实现：
FileChannel 从文件中读写数据。
DatagramChannel 能通过UDP读写网络中的数据。
SocketChannel 能通过TCP读写网络中的数据。
ServerSocketChannel可以监听新进来的TCP连接，像Web服务器那样。对每一个新进来的连接都会创建一个SocketChannel。

Channel使用示例

import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;

/**
 * Created by chenglin on 2018/7/31.
 */
public class Test {

    public static void main(String[] args) {
        RandomAccessFile aFile = null;
        try {
            aFile = new RandomAccessFile("F:/nio-data.txt", "rw");

            FileChannel inChannel = aFile.getChannel();

            ByteBuffer buf = ByteBuffer.allocate(48);

            int bytesRead = inChannel.read(buf);

            while (bytesRead != -1) {
                System.out.println("Read " + bytesRead);
                buf.flip();
                while(buf.hasRemaining()){
                    System.out.print((char) buf.get());
                }
                buf.clear();
                bytesRead = inChannel.read(buf);
            }
            aFile.close();

        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

Buffer

Buffer的定义

  在计算机领域，缓冲器指的是缓冲寄存器，它分输入缓冲器和输出缓冲器两种。前者的作用是将外设送来的数据暂时存放，以便处理器将它取走；后者的作用是用来暂时存放处理器送往外设的数据。通常我们也叫它缓存，缓冲区本质上是一块可以写入数据，然后可以从中读取数据的内存。这块内存被包装成NIO Buffer对象，并提供了一组方法，用来方便的访问该块内存。

Buffer的实现

[2] 以下是Java NIO里关键的Buffer实现：
ByteBuffer
CharBuffer
DoubleBuffer
FloatBuffer
IntBuffer
LongBuffer
ShortBuffer
还有个 MappedByteBuffer，用于表示内存映射文件，

Buffer读写数据

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

Buffer的结构

capacity

  作为一个内存块，Buffer有一个固定的大小值，也叫“capacity”.你只能往里写capacity个byte、long，char等类型。一旦Buffer满了，需要将其清空（通过读数据或者清除数据）才能继续写数据往里写数据。

position

  当你写数据到Buffer中时，position表示当前的位置。初始的position值为0.当一个byte、long等数据写到Buffer后， position会向前移动到下一个可插入数据的Buffer单元。position最大可为capacity – 1；当读取数据时，也是从某个特定位置读。当将Buffer从写模式切换到读模式，position会被重置为0. 当从Buffer的position处读取数据时，position向前移动到下一个可读的位置。

limit

  在写模式下，Buffer的limit表示你最多能往Buffer里写多少数据。写模式下，limit等于Buffer的capacity。当切换Buffer到读模式时，limit表示你最多能读到多少数据。因此，当切换Buffer到读模式时，limit会被设置成写模式下的position值。换句话说，你能读到之前写入的所有数据（limit被设置成已写数据的数量，这个值在写模式下就是position）

Buffer的使用示例

import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;

/**
 * Created by chenglin on 2018/7/31.
 */
public class Test {

    public static void main(String[] args) {
        RandomAccessFile aFile = null;
        try {
            aFile = new RandomAccessFile("F:/nio-data.txt", "rw");

            // 获取一个channel
            FileChannel inChannel = aFile.getChannel();

            // Buffer的分配
            ByteBuffer buf = ByteBuffer.allocate(48);

            // 向Buffer中写数据
            int bytesRead = inChannel.read(buf);

            while (bytesRead != -1) {
                System.out.println("Read " + bytesRead);

                // flip将Buffer从写模式切换到读模式。调用flip()会将position设回0，并将limit设置成之前position的值。
                buf.flip();

                // 从Buffer中读取数据
                while(buf.hasRemaining()){
                    System.out.print((char) buf.get());
                }
                // 一旦读完Buffer中的数据，需要让Buffer准备好再次被写入。可以通过clear()或compact()方法来完成。
                buf.clear();
                bytesRead = inChannel.read(buf);

                // 将position设回0，所以你可以重读Buffer中的所有数据。
                // limit保持不变，仍然表示能从Buffer中读取多少个元素（byte、char等）。
                buf.rewind();

                // mark标记Buffer中的一个特定position。之后可以通过调用Buffer.reset()方法恢复到这个position
                buf.mark();
                buf.reset();
            }
            aFile.close();

        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

Scatter/Gather

  SelectorJava NIO开始支持scatter/gather，scatter/gather用于描述从Channel（Channel在中文经常翻译为通道）中读取或者写入到Channel的操作。

• 分散（scatter）从Channel中读取是指在读操作时将读取的数据写入多个buffer中。因此，Channel将从Channel中读取的数据“分散（scatter）”到多个Buffer中。
• 聚集（gather）写入Channel是指在写操作时将多个buffer的数据写入同一个Channel，因此，Channel 将多个Buffer中的数据“聚集（gather）”后发送到Channel。

  scatter / gather经常用于需要将传输的数据分开处理的场合，例如传输一个由消息头和消息体组成的消息，你可能会将消息体和消息头分散到不同的buffer中，这样你可以方便的处理消息头和消息体。

Scattering Reads

Scattering Reads是指数据从一个channel读取到多个buffer中。

ByteBuffer header = ByteBuffer.allocate(128);
ByteBuffer body   = ByteBuffer.allocate(1024);

ByteBuffer[] bufferArray = { header, body };

channel.read(bufferArray);

  注意buffer首先被插入到数组，然后再将数组作为channel.read() 的输入参数。read()方法按照buffer在数组中的顺序将从channel中读取的数据写入到buffer，当一个buffer被写满后，channel紧接着向另一个buffer中写。
  Scattering Reads在移动下一个buffer前，必须填满当前的buffer，这也意味着它不适用于动态消息(消息大小不固定)。换句话说，如果存在消息头和消息体，消息头必须完成填充（例如 128byte），Scattering Reads才能正常工作。

Gathering Writes

Gathering Writes是指数据从多个buffer写入到同一个channel。

ByteBuffer header = ByteBuffer.allocate(128);
ByteBuffer body   = ByteBuffer.allocate(1024);

//write data into buffers

ByteBuffer[] bufferArray = { header, body };

channel.write(bufferArray);

  buffers数组是write()方法的入参，write()方法会按照buffer在数组中的顺序，将数据写入到channel，注意只有position和limit之间的数据才会被写入。因此，如果一个buffer的容量为128byte，但是仅仅包含58byte的数据，那么这58byte的数据将被写入到channel中。因此与Scattering Reads相反，Gathering Writes能较好的处理动态消息。

通道之间的数据传输

在Java NIO中，如果两个通道中有一个是FileChannel，那你可以直接将数据从一个channel传输到另外一个channel。

代码段

{
    RandomAccessFile fromFile = new RandomAccessFile("fromFile.txt", "rw");
    FileChannel      fromChannel = fromFile.getChannel();

    RandomAccessFile toFile = new RandomAccessFile("toFile.txt", "rw");
    FileChannel      toChannel = toFile.getChannel();

    long position = 0;
    long count = fromChannel.size();

    // ransferFrom()方法可以将数据从源通道传输到FileChannel中
    toChannel.transferFrom(position, count, fromChannel);
}

{
    RandomAccessFile fromFile = new RandomAccessFile("fromFile.txt", "rw");
    FileChannel      fromChannel = fromFile.getChannel();

    RandomAccessFile toFile = new RandomAccessFile("toFile.txt", "rw");
    FileChannel      toChannel = toFile.getChannel();

    long position = 0;
    long count = fromChannel.size();

    // transferTo()方法将数据从FileChannel传输到其他的channel中。
    fromChannel.transferTo(position, count, toChannel);
}

Selector

Selector的定义

  Selector（选择器）是Java NIO中能够检测一到多个NIO通道，并能够知晓通道是否为诸如读写事件做好准备的组件。这样，一个单独的线程可以管理多个channel，从而管理多个网络连接。
  如果你的应用打开了多个连接（通道），但每个连接的流量都很低，使用Selector就会很方便。
  要使用Selector，得向Selector注册Channel，然后调用它的select()方法。这个方法会一直阻塞到某个注册的通道有事件就绪。一旦这个方法返回，线程就可以处理这些事件，事件的例子有如新连接进来，数据接收等。

为什么使用Selector?

  仅用单个线程来处理多个Channels的好处是，只需要更少的线程来处理通道。事实上，可以只用一个线程处理所有的通道。对于操作系统来说，线程之间上下文切换的开销很大，而且每个线程都要占用系统的一些资源（如内存）。因此，使用的线程越少越好。

Selector的结构

使用示例代码段

public static void main(String[] args) {
    // 通过调用Selector.open()方法创建一个Selector
    Selector selector = Selector.open();

    // 为了将Channel和Selector配合使用，必须将channel注册到selector上。通过SelectableChannel.register()方法来实现
    // 与Selector一起使用时，Channel必须处于非阻塞模式下。
    // 这意味着不能将FileChannel与Selector一起使用，因为FileChannel不能切换到非阻塞模式。而套接字通道都可以。
    channel.configureBlocking(false);

    // register的第二个参数有4个值：OP_CONNECT、OP_ACCEPT、OP_READ、OP_WRITE
    // 如果想注册多个事件，可以用“位或”操作符：int interestSet = SelectionKey.OP_READ | SelectionKey.OP_WRITE;
    SelectionKey key = channel.register(selector, SelectionKey.OP_READ);

    // SelectionKey里面就包含了所有后面需要用的的对象和方法，具体定义可以看下一节的源码
    while(true) {

        // select()方法返回的int值表示有多少通道已经就绪。
        int readyChannels = selector.select();

        // 如果没有通道就绪，则继续轮询
        if(readyChannels == 0) continue;

        // 调用selector的selectedKeys()方法，访问“已选择键集（selected key set）”中的就绪通道。
        Set selectedKeys = selector.selectedKeys();

        // 通过迭代器遍历已就绪的通道
        Iterator keyIterator = selectedKeys.iterator();
        while(keyIterator.hasNext()) {
            SelectionKey key = keyIterator.next();
            if(key.isAcceptable()) {
                // a connection was accepted by a ServerSocketChannel.
            } else if (key.isConnectable()) {
                // a connection was established with a remote server.
            } else if (key.isReadable()) {
                // a channel is ready for reading
            } else if (key.isWritable()) {
                // a channel is ready for writing
            }
            keyIterator.remove();
        }
    }

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

    // 用完Selector后调用其close()方法会关闭该Selector，且使注册到该Selector上的所有SelectionKey实例无效。
    // 通道本身并不会关闭。
    selector.close();

}

SelectionKey定义

package com.xunsiya;

import java.nio.channels.SelectableChannel;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;

public abstract class SelectionKey {

    protected SelectionKey() { }

    // -- Channel and selector operations --

    /**
     * @return  This key's channel
     */
    public abstract SelectableChannel channel();

    /**
     * @return  This key's selector
     */
    public abstract Selector selector();

    /**
     * @return  <tt>true</tt> if, and only if, this key is valid
     */
    public abstract boolean isValid();

    public abstract void cancel();

    // -- Operation-set accessors --

    /**
     * @return  This key's interest set
     * Set：OP_CONNECT、OP_ACCEPT、OP_READ、OP_WRITE
     */
    public abstract int interestOps();

    /**
     * @param  ops  The new interest set
     * @return  This selection key
     */
    public abstract SelectionKey interestOps(int ops);

    /**
     * @return  This key's ready-operation set
     */
    public abstract int readyOps();

    /**
     * Operation-set bit for read operations.
     */
    public static final int OP_READ = 1 << 0;

    /**
     * Operation-set bit for write operations.
     */
    public static final int OP_WRITE = 1 << 2;

    /**
     * Operation-set bit for socket-connect operations.
     */
    public static final int OP_CONNECT = 1 << 3;

    /**
     * Operation-set bit for socket-accept operations.
     */
    public static final int OP_ACCEPT = 1 << 4;

    /**
     * Tests whether this key's channel is ready for reading.
     */
    public final boolean isReadable() {
        return (readyOps() & OP_READ) != 0;
    }

    /**
     * Tests whether this key's channel is ready for writing.
     */
    public final boolean isWritable() {
        return (readyOps() & OP_WRITE) != 0;
    }

    /**
     * Tests whether this key's channel has either finished, or failed to finish, its socket-connection operation.
     */
    public final boolean isConnectable() {
        return (readyOps() & OP_CONNECT) != 0;
    }

    /**
     * Tests whether this key's channel is ready to accept a new socket connection.
     */
    public final boolean isAcceptable() {
        return (readyOps() & OP_ACCEPT) != 0;
    }

    // -- Attachments -- 附加的对象

    private volatile Object attachment = null;

    private static final AtomicReferenceFieldUpdater<SelectionKey,Object>
        attachmentUpdater = AtomicReferenceFieldUpdater.newUpdater(
            SelectionKey.class, Object.class, "attachment"
        );

    /**
     * Attaches the given object to this key.
     */
    public final Object attach(Object ob) {
        return attachmentUpdater.getAndSet(this, ob);
    }

    /**
     * Retrieves the current attachment.
     */
    public final Object attachment() {
        return attachment;
    }

}

非Async的Channel

FileChannel

特性

• Java NIO中的FileChannel是一个连接到文件的通道。可以通过文件通道读写文件。
• FileChannel无法设置为非阻塞模式，它总是运行在阻塞模式下。

使用

// 打开FileChannel
RandomAccessFile aFile = new RandomAccessFile("data/nio-data.txt", "rw");
FileChannel inChannel = aFile.getChannel();

// 从FileChannel读取数据
ByteBuffer buf = ByteBuffer.allocate(48);
int bytesRead = inChannel.read(buf);

// 向FileChannel写数据
String newData = "New String to write to file..." + System.currentTimeMillis();

ByteBuffer buf = ByteBuffer.allocate(48);
buf.clear();
buf.put(newData.getBytes());

buf.flip();

while(buf.hasRemaining()) {
    channel.write(buf);
}

// 关闭FileChannel
channel.close();

// 其他一些方法使用是看API说明即可，比如：
// position：返回当前位置或指定操作位置
long pos = channel.position();
channel.position(pos +123);

// size：返回该实例所关联文件的大小 
long fileSize = channel.size();

// truncate：可以使用FileChannel.truncate()方法截取一个文件。截取文件时，文件将中指定长度后面的部分将被删除 
channel.truncate(1024);

// force：将通道里尚未写入磁盘的数据强制写到磁盘上
// force()方法有一个boolean类型的参数，指明是否同时将文件元数据（权限信息等）写到磁盘上
channel.force(true);

SocketChannel

特性

  Java NIO中的SocketChannel是一个连接到TCP网络套接字的通道。也可以手动设置 SocketChannel 为非阻塞模式（non-blocking mode）;设置之后，就可以在异步模式下调用connect(), read()和write()了，非阻塞模式与选择器搭配会工作的更好，通过将一或多个SocketChannel注册到Selector，可以询问选择器哪个通道已经准备好了读取，写入等。
  通常可以通过以下2种方式创建SocketChannel：

• 打开一个SocketChannel并连接到互联网上的某台服务器。
• 一个新连接到达ServerSocketChannel时，会创建一个SocketChannel。

使用

// 打开 SocketChannel
SocketChannel socketChannel = SocketChannel.open();
socketChannel.connect(new InetSocketAddress("http://xxx.com", 80));

// 从 SocketChannel 读取数据
ByteBuffer buf = ByteBuffer.allocate(48);
int bytesRead = socketChannel.read(buf);

// 写入 SocketChannel
String newData = "New String to write to file..." + System.currentTimeMillis();

ByteBuffer buf = ByteBuffer.allocate(48);
buf.clear();
buf.put(newData.getBytes());

buf.flip();

while(buf.hasRemaining()) {
    channel.write(buf);
}

// 关闭 SocketChannel
socketChannel.close();

// 设置 SocketChannel 为非阻塞模式（non-blocking mode）
socketChannel.configureBlocking(false);
socketChannel.connect(new InetSocketAddress("http://xxx.com", 80));

while(! socketChannel.finishConnect() ){
    //wait, or do something else...
}

ServerSocketChannel

特性

  Java NIO中的 ServerSocketChannel 是一个可以监听新进来的TCP连接的通道, 就像标准IO中的ServerSocket一样。ServerSocketChannel类在 java.nio.channels包中。

使用

// 打开 ServerSocketChannel
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();

// 绑定端口
serverSocketChannel.socket().bind(new InetSocketAddress(9999));

// 设置成非阻塞模式，在非阻塞模式下，accept() 方法会立刻返回
serverSocketChannel.configureBlocking(false);

// 监听新进来的连接
while(true){
    SocketChannel socketChannel =
            serverSocketChannel.accept();

    // 如果还没有新进来的连接,返回的将是null
    if(socketChannel != null){
        //do something with socketChannel...
    }
}

// 关闭 ServerSocketChannel
serverSocketChannel.close();

DatagramChannel

特性

  Java NIO中的DatagramChannel是一个能收发UDP包的通道。因为UDP是无连接的网络协议，所以不能像其它通道那样读取和写入。它发送和接收的是数据包。

使用

// 打开 DatagramChannel
DatagramChannel channel = DatagramChannel.open();
channel.socket().bind(new InetSocketAddress(9999));

// 接收数据
ByteBuffer buf = ByteBuffer.allocate(48);
buf.clear();
channel.receive(buf);

// 发送数据
String newData = "New String to write to file..." + System.currentTimeMillis();

ByteBuffer buf = ByteBuffer.allocate(48);
buf.clear();
buf.put(newData.getBytes());
buf.flip();

int bytesSent = channel.send(buf, new InetSocketAddress("xxx.com", 80));

// 关闭 DatagramChannel
channel.close();

// 可以将DatagramChannel“连接”到网络中的特定地址进行读写；就像在用传统的通道一样。只是在数据传送方面没有任何保证。
channel.connect(new InetSocketAddress("xxx.com", 80));
int bytesRead = channel.read(buf);
int bytesWritten = channel.write(but);

Async的Channel

实际这里的Channel提供的就是AIO的服务支撑，所谓的AIO也可以理解为NIO2.0

AsynchronousFileChannel

An asynchronous channel for reading, writing, and manipulating a file.@since 1.7

AsynchronousServerSocketChannel

An asynchronous channel for stream-oriented listening sockets.@since 1.7

AsynchronousSocketChannel

An asynchronous channel for stream-oriented connecting sockets.@since 1.7

AIO的应用示例

服务端代码

import java.io.IOException;
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.Date;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * By Chenglin
 */
public class AIODemoServer {
 
    public final static int PORT = 8088;
    public final static String IP = "127.0.0.1";
 
    // 声明一个Channel
    private AsynchronousServerSocketChannel socketChannel = null;
    
    public AIODemoServer(){
        try {
            // 产生一个异步通道绑定到本机的8088端口
            socketChannel = AsynchronousServerSocketChannel.open().bind(new InetSocketAddress(IP, PORT));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 使用这个socketChannel来进行客户端的消息接收和处理
     */
    public void start(){
        System.out.println("Server start port "+PORT);
        
        // 注册事件和事件完成后的处理器，CompletionHandler就是事件完成后的处理器（可自己实现具体的操作）
        socketChannel.accept(null, new CompletionHandler<AsynchronousSocketChannel,Object>(){
 
            final ByteBuffer buffer = ByteBuffer.allocate(1024);
            
            @Override
            public void completed(AsynchronousSocketChannel result,Object attachment) {
                
                Future<Integer> writeResult = null;
                
                try{
                    buffer.clear();

                    // 从缓冲区读取数据
                    result.read(buffer).get(100,TimeUnit.SECONDS);

                    System.out.println("[" + new Date() + "] Server received message : "+ new String(buffer.array()));

                    // 将数据回复给客户端
                    buffer.flip();
                    writeResult = result.write(buffer);

                    System.out.println("[" + new Date() + "] Server send message back : "+ new String(buffer.array()));

                }catch(InterruptedException | ExecutionException | TimeoutException e){
                    e.printStackTrace();
                }finally{
                    socketChannel.accept(null,this);
                    try {
                        writeResult.get();
                        result.close();
                    } catch (InterruptedException | ExecutionException e) {
                        e.printStackTrace();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
                
            }
 
            @Override
            public void failed(Throwable exc, Object attachment) {
                System.out.println("Server failed:" + exc);
            }
            
        });
    }

    /**
     * Main方法（启动服务，然后死循环）
     * @param args
     */
    public static void main(String[] args) {
        new AIODemoServer().start();
        while(true){
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
 
}

客户端代码

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.util.Date;

/**
 * By Chenglin
 */
public class AIODemoClient {

    /**
     * Main方法
     * @param args
     * @throws IOException
     */
    public static void main(String[] args) throws IOException {

        String message = "Hello world";

        // 声明一个Channel
        final AsynchronousSocketChannel socketChannel = AsynchronousSocketChannel.open();

        // 定义Server的IP和端口
        InetSocketAddress serverAddress = new InetSocketAddress("127.0.0.1",8088);

        // 定义一个事件完成后的处理器（可自己实现具体的操作）
        CompletionHandler<Void, ? super Object> handler = new CompletionHandler<Void,Object>(){
 
            @Override
            public void completed(Void result, Object attachment) {

                System.out.println("[" + new Date() + "] Client send message : " + message);

                // 写一个Hello world，然后读取对端的返回结果
                socketChannel.write(ByteBuffer.wrap(message.getBytes()),null,
                        new CompletionHandler<Integer,Object>(){
 
                            @Override
                            public void completed(Integer result,
                                    Object attachment) {
                                final ByteBuffer buffer = ByteBuffer.allocate(1024);
                                socketChannel.read(buffer, buffer, new CompletionHandler<Integer,ByteBuffer>(){
 
                                    @Override
                                    public void completed(Integer result,
                                            ByteBuffer attachment) {
                                        buffer.flip();
                                        System.out.println("[" + new Date() + "] Client received back message : " + new String(buffer.array()));
                                        try {
                                            socketChannel.close();
                                        } catch (IOException e) {
                                            e.printStackTrace();
                                        }
                                    }
 
                                    @Override
                                    public void failed(Throwable exc,
                                            ByteBuffer attachment) {
                                    }
                                    
                                });
                            }
 
                            @Override
                            public void failed(Throwable exc, Object attachment) {
                            }
                    
                });
            }
 
            @Override
            public void failed(Throwable exc, Object attachment) {
            }
            
        };
        
        socketChannel.connect(serverAddress, null, handler);
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
 
}

运行结果

"C:\Program Files\Java\jdk1.8.0_131\bin\java" "......" AIODemoServer
Server start port 8088
[Wed Aug 01 11:28:55 CST 2018] Server received message : Hello world
[Wed Aug 01 11:28:55 CST 2018] Server send message back : Hello world

"C:\Program Files\Java\jdk1.8.0_131\bin\java" "......" AIODemoClient
[Wed Aug 01 11:28:55 CST 2018] Client send message : Hello world
[Wed Aug 01 11:28:55 CST 2018] Client received back message : Hello world

参考：非阻塞式服务器

  一个非阻塞式服务器需要时不时检查输入的消息来判断是否有任何的新的完整的消息发送过来。服务器可能会在一个或多个完整消息发来之前就检查了多次。检查一次是不够的。同样，一个非阻塞式服务器需要时不时检查是否有任何数据需要写入。如果有，服务器需要检查是否有任何相应的连接准备好将该数据写入它们。只有在第一次排队消息时才检查是不够的，因为消息可能被部分写入。所以非阻塞服务器最终都需要定期执行的三个“管道”（pipelines），大致的处理流程如下图（来源于网络，仅供参考）：

• 读取管道（The read pipeline），用于检查是否有新数据从开放连接进来的。
• 处理管道(The process pipeline)，用于所有任何完整消息。
• 写入管道（The write pipeline），用于检查是否可以将任何传出的消息写入任何打开的连接。