Java NIO 详解---NIO中的异步网络IO

前面的例子都是关于如何通过NIO操作文件读写的，我们知道BIO中的Socket、ServerSocket提供了网络通信的能力，在NIO中也有对应的模块提供了这种能力，并且具有更加强大的功能—通过异步非阻塞的数据读写实现一个线程监听多个连接的能力。

1）异步IO

所谓的异步IO是一种没有阻塞读写数据的方法。通常情况下，代码在调用read()方法时程序会阻塞直到又可以读取的数据；同样代码在写入数据的时候，代码会阻塞直到数据写入完成。而异步IO不会有这种阻塞，相反应用程序将注册自己感兴趣的IO事件—可读数据的到达、写入数据完成、新的连接的到来，当这些事情发生的时候系统将会通知应用程序；这样的好处之一就在于可以使用一个线程操操作多个IO而不用像传统程序那样同步的轮询或需要使用多个线程来处理。

2）Selector

Selector(选择器)是NIO中能够监听一到多个通道，并且知道这些通道是否为读写做好准备的组件，这样一个线程可以通过管理多个Channel，进而管理多个网络连接。使用一个线程管理多个网络连接的好处在于可以避免线程间切换的开销。

下面示范如何以一个Selector管理Channel。首先是Selector的建立：

//通过静态的open()方法得到一个Selector
Selector selector = Selector.open();

然后是向Selector注册一个ServerSocketChannel并监听连接事件

//对于监听的端口打开一个ServerSocketChannel
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
//注册到Selector的Channel必须设置为非阻塞模式,否则实现不了异步IO
serverSocketChannel.configureBlocking(false);
ServerSocket serverSocket = serverSocketChannel.socket();
InetSocketAddress address = new InetSocketAddress(8410);
serverSocket.bind(address);
//第二个参数是表明这个Channel感兴趣的事件
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

与Selector同时使用的Channel必须处于非阻塞模式，这意味着FileChannel不能用于Selector，因为它不能切换到非阻塞通道；而套接字通道都是可以的。

register的第二个参数表明了该Channel感兴趣的事件，具体的事件分为四个类型 1.Connect 2.Accept 3.Read 4.Write ，具体来说某个channel成功连接到另一个服务器称为“连接就绪”。一个server socket channel准备好接收新进入的连接称为“接收就绪”。一个有数据可读的通道可以说是“读就绪”。等待写数据的通道可以说是“写就绪”。这些事件可以用SelectionKey的四个常量来表示：

SelectionKey.OP_CONNECT
SelectionKey.OP_ACCEPT
SelectionKey.OP_READ
SelectionKey.OP_WRITE

上面的Channel只是注册了一个事件，但实际上是可以同时注册多个事件的，比如可以像下面这样同时注册”接收就绪”和”读就绪”两个事件:

//使用"|"连接同时注册多个事件
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT|SelectionKey.OP_READ);

3) SelectionKey

上面向Selector注册Channel后返回了一个SelectionKey对象，这个对象包含了一些很有用的信息集：

interest集合
ready集合
Channel
Selector

interest集合即上面Channel注册时添加的感兴趣的事件集合，我们可以通过调用SelectionKey 的interestOps()方法得到一个int数字，然后通过&位操作来确定具体有哪些感兴趣的集合：

int interestSet = key.interestOps();
//是否包含ACCEPT事件
boolean isInterestedInAccept = (interestSet & SelectionKey.OP_ACCEPT) == SelectionKey.OP_ACCEPT;
//是否包含CONNECT事件
boolean isInterestedInConnect = (interestSet & SelectionKey.OP_CONNECT) == SelectionKey.OP_CONNECT;
boolean isInterestedInRead = (interestSet & SelectionKey.OP_READ) == SelectionKey.OP_READ;
boolean isInterestedInWrite = (interestSet & SelectionKey.OP_WRITE) == SelectionKey.OP_WRITE;

ready集合表明该Selector上已经就绪的事件，可以通过key.readyOps()获得一个数字，然后通过上面同样的方式拿到就绪的集合；但是，也可以使用下面这些更加简洁的方法判断:

//四个返回boolean值的方法，可以用于判断目前Selector上有哪些事件已经就绪
selectionKey.isAcceptable();
selectionKey.isConnectable();
selectionKey.isReadable();
selectionKey.isWritable();

可以很简单的拿到这个SelectinKey关联的Selector和Channel，如下所示：

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

3）监听Selector选择通道

当向Selector注册了几个Channel之后，就可以调用几个重载的select()方法来检测是否有通道已经就绪了。具体的来说，Selector的select()方法有以下三种形式：

int select()
int select(long timeout)
int selectNow()

第一个方法会阻塞直到至少有一个通道就绪然后返回；第二个方法和第一个方法类似但不会一直阻塞而是至多会阻塞timeout时间；第三个方法不会阻塞，无论有无就绪的通道都会立即返回，如果没有就绪的通道会返回0。这些方法返回的int值表明该Selector上就绪通道的数量，准确的来说是自上次调用select()方法后有多少通道变成就绪状态。如果调用select()方法，因为有一个通道变成就绪状态，返回了1，若再次调用select()方法，如果另一个通道就绪了，它会再次返回1。如果对第一个就绪的channel没有做任何操作，现在就有两个就绪的通道，但在每次select()方法调用之间，只有一个通道就绪了。

如果调用select()方法表明至少有一个通道就绪了，那么就可以通过selector.selectedKeys()方法来获得具体就绪的通道，这个方法的返回值是Set。如上面所介绍的我们可以很方便的通过SelectionKey找到就绪的事件以及对应的Channel，下面的代码示例了如何遍历这个Set：

Set<SelectionKey> selectionKeySet  = selector.selectedKeys();
Iterator<SelectionKey> iterator = selectionKeySet.iterator();
while (iterator.hasNext()){
SelectionKey selectionKey = iterator.next();
if(selectionKey.isAcceptable()){
// a connection was accepted by a ServerSocketChannel.
}else if(selectionKey.isConnectable()){
// a connection was established with a remote server.
}else if(selectionKey.isWritable()){
// a channel is ready for writing
}else if(selectionKey.isReadable()){
// a channel is ready for reading
}
iterator.remove();
}

注意末尾的remove()方法，当处理完一个SelectionKey之后，必须手动的将其从Set中移除，Selector本身不会进行这个工作，所以需要我们手动移除避免下一次重复处理。

4）ServerSocketChannel

其实从上面的代码中我们已经看到了，ServerSocketChannel和ServerSocket所起的作用是一致的，都是用来监听tcp连接的；值得注意的就是ServerSocketChannel是可以设置为非阻塞模式的，这时候它的accept()方法在没有连接进入的情况下总是返回null。下面的代码示例了ServerSocketChannel的基本用法：

//ServerSocketChannel对象通过静态方法获取
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
//具体的端口绑定操作还是通过关联的ServerSocket实现
ServerSocket ss = serverSocketChannel.socket();
InetSocketAddress address = new InetSocketAddress(8410);
ss.bind(address);
//ServerSocketChannel可以被设置成非阻塞的模式,这是和Selector配合使用的基础
serverSocketChannel.configureBlocking(false);
while (true){
//accept()方法用于监听进来的连接,如果被设置为非阻塞模式,那么当没有连接时总是返回null
SocketChannel socketChannel = serverSocketChannel.accept();
if (socketChannel != null) {
//do something with socketChannel...
}
}

5) SocketChannel

Java NIO中的SocketChannel是一个连接到TCP网络套接字的通道,和Socket是类似的。可以通过以下2种方式创建SocketChannel：

打开一个SocketChannel并连接到互联网上的某台服务器。

SocketChannel socketChannel = SocketChannel.open();
socketChannel.connect(new InetSocketAddress("localhost",80));

一个新连接到达ServerSocketChannel时，会创建一个SocketChannel。如上面介绍ServerSocketChannel的代码所示

SocketChannel的数据读写和前面介绍的FileChannel没有什么不同，都是需要借助Buffer；值得注意的是SocketChannel是可以工作在非阻塞模式下的，这时候的read()、write()方法都会直接返回，这种模式主要是为了配合Selector来实现异步非阻塞IO。

最后是一个总的示例：

//测试一个线程同时监听多个端口并且同时处理多个连接
public static void testSocketNIO() {

//需要监听的端口list
List<Integer> portList = new ArrayList<Integer>();
portList.add(8410);
portList.add(8411);
portList.add(8412);

//注册端口监听事件,并使用异步IO形式使用一个线程监听多个端口并处理多个连接
go(portList);

}

public static void go(List<Integer> portList) {
//进行监听的选择器
Selector selector = null;
try {
selector = Selector.open();
} catch (IOException e) {
return;
}
//每个端口开一个ServerSocketChannel进行监听
for (int port : portList) {
try {
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.configureBlocking(false);//配置成非阻塞模式
ServerSocket serverSocket = serverSocketChannel.socket();
InetSocketAddress address = new InetSocketAddress(port);
serverSocket.bind(address);
//注册监听新连接就绪事件
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
} catch (IOException e) {

}
}
//监听具体事件
while (true) {
int select = 0;
try {
select = selector.select();
} catch (IOException e) {

}
if (select == 0) {
continue;
}
//拿到具体就绪的Channel进行处理
Set<SelectionKey> selectionKeyList = selector.selectedKeys();
Iterator<SelectionKey> iterator = selectionKeyList.iterator();
while (iterator.hasNext()) {
SelectionKey selectionKey = iterator.next();
if (selectionKey.isAcceptable()) { //处理新连接进入事件
ServerSocketChannel serverSocketChannel = (ServerSocketChannel) selectionKey.channel();
SocketChannel socketChannel = null;
try {
socketChannel = serverSocketChannel.accept();
socketChannel.configureBlocking(false);//新连接同时设置为非阻塞,同样使用上面的Selector进行监听
System.out.println("accept new Connection " + socketChannel.getRemoteAddress());
socketChannel.register(selector, SelectionKey.OP_READ); //监听数据传入事件
} catch (IOException e) {
}
} else if (selectionKey.isReadable()) {
SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
ByteBuffer result = ByteBuffer.allocate(1024);
ByteBuffer byteBuffer = ByteBuffer.allocate(3);
try {
result.clear();
while (true) {
byteBuffer.clear();
int num = socketChannel.read(byteBuffer);
if (num == -1) {
break;
}
byteBuffer.flip();
result.put(byteBuffer);
}
result.flip();
Charset charset = Charset.forName("UTF-8");
CharsetDecoder decoder = charset.newDecoder();
System.out.println("receive " + decoder.decode(result).toString() + " from " + socketChannel.getRemoteAddress());
} catch (IOException e) {

}
}
//处理完这个SelectionKey之后就需要将其移除掉
iterator.remove();
}

}
}