Java NIO (三)

在上一篇中介绍了缓冲区的原理，下面来介绍NIO中另一个核心对象选择器（Selector）以及NIO的原理。

在Client/Server模型中，Server往往需要同时处理大量来自Client的访问请求，因此Server端需采用支持高并发访问的架构。一种简单而又直接的解决方案是“one-thread-per-connection”。这是一种基于阻塞式I/O的多线程模型。在该模型中，Server为每个Client连接创建一个处理线程，每个处理线程阻塞式等待可能达到的数据，一旦数据到达，则立即处理请求、返回处理结果并再次进入等待状态。由于每个Client连接有一个单独的处理线程为其服务，因此可保证良好的响应时间。但当系统负载增大（并发请求增多）时，Server端需要的线程数会增加，这将成为系统扩展的瓶颈所在。

Java NIO (三)

Java NIO不但引入了全新的高效的I/O机制，同时引入了基于Reactor设计模式的多路复用异步模式。NIO包中主要包含以下几种核心对象。

* Channel(通道)：NIO把它支持的I/O对象抽象为Channel。它模拟了通信连接，类似于原I/O中的流（Stream），用户可以通过它读取和写入数据。实例类有SocketChannel、ServerSocketChannel、DatagramChannel、FileChannel等。

* Buffer(缓冲区)：Buffer是一块连续的内存区域，一般作为Channel收发数据的载体出现。所有数据都通过Buffer对象来处理。

* Selector(选择器)：Selector类提供了监控一个和多个通道当前状态的机制。只要Channel向Selector注册了某种特定的事件，Selector就会监听这些事件是否会发生，一旦发生某个事件，便会通知对应的Channel。使用选择器，借助单一线程，就可对数量庞大的活动I/O通道实施监控和维护。

Java NIO (三)

Java NIO的服务端只需启动一个专门的线程来处理所有的IO事件，这种通信模型是怎么实现的呢？Java NIO采用了双向通道（Channel）进行数据传输，而不是单向的流（Stream），在通道上可以注册我们感兴趣的事件。一共有以下四种事件：

事件名	对应值
服务端接收客户端连接事件	SelectionKey.OP_ACCEPT(16)
客户端连接服务端事件	SelectionKey.OP_CONNECT(8)
读事件	SelectionKey.OP_READ(1)
写事件	SelectionKey.OP_WRITE(4)

服务端和客户端各自维护一个管理通道的对象，我们称之为Selector，该对象能检测一个或多个通道 (Channel) 上的事件。我们以服务端为例，如果服务端的Selector上注册了读事件，某时刻客户端给服务端发送了一些数据，阻塞I/O这时会调用read()方法阻塞地读取数据，而NIO的服务端会在Selector中注册一个读事件，然后服务端的处理线程会轮询地访问Selector，如果发现有读事件到达，则处理这些事件，如果没有则处理线程会一直阻塞直到读事件到达为止。

为了更好地理解java NIO，下面贴出服务端和客户端的简单代码实现。

服务端：

[java] view plain copy

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
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;
/**
* NIO服务端
*
*/
public class NIOServer {
private Selector selector;
private ByteBuffer echoBuffer = ByteBuffer.allocate(1024);
/**
* 获得一个ServerSocket通道，并对该通道做一些初始化的工作
*
* @param port
* @throws IOException
*/
public void initServer(int port) throws IOException {
selector = Selector.open();
// 获得一个ServerSocket通道，并设置为非阻塞
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.configureBlocking(false);
// 将通道对应的ServerSocket绑定到port端口
ServerSocket serverSocket = serverSocketChannel.socket();
InetSocketAddress address = new InetSocketAddress(port);
serverSocket.bind(address);
// 将通道管理器和该通道绑定，并为该通道注册SelectionKey.OP_ACCEPT事件。
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
}
/**
* 循环监听selector上是否有需要处理的事件，如果有，则进行处理
*
* @throws IOException
*/
public void listen() throws IOException {
System.out.println("服务端启动成功！");
// 循环访问selector
while (true) {
// //当注册的事件到达时，方法返回；否则,该方法会一直阻塞
selector.select();
// 返回发生了事件的SelectionKey对象的一个集合
Set<SelectionKey> selectedKeys = selector.selectedKeys();
Iterator<SelectionKey> it = selectedKeys.iterator();
while (it.hasNext()) {
SelectionKey key = (SelectionKey) it.next();
it.remove(); // 删除已选的key以防重复处理
// 客户端请求连接事件
if ((key.readyOps() & SelectionKey.OP_ACCEPT) == SelectionKey.OP_ACCEPT) {
ServerSocketChannel server = (ServerSocketChannel) key
.channel();
SocketChannel channel = server.accept(); // 获得和客户端连接的通道
channel.configureBlocking(false);// 设置成非阻塞
// 在和客户端连接成功之后，为了可以接收到客户端的信息，需要给通道注册读事件。
channel.register(this.selector, SelectionKey.OP_READ);
} else if ((key.readyOps() & SelectionKey.OP_READ) == SelectionKey.OP_READ) {
// 得到可读事件发生的Socket通道
SocketChannel channel = (SocketChannel) key.channel();
while (true) {
echoBuffer.clear();
int r = channel.read(echoBuffer);
// 判断拷贝是否完成
if (r <= 0) {
break;
}
echoBuffer.flip();
channel.write(echoBuffer); // 将消息回送给客户端
}
byte[] data = echoBuffer.array();
System.out.println("服务器收到并返回信息：" + new String(data).trim());
}
}
}
}
/**
* 启动服务端测试
*
* @throws IOException
*/
public static void main(String[] args) throws IOException {
NIOServer server = new NIOServer();
server.initServer(8000);
server.listen();
}
}

客户端：

[java] view plain copy

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;
/**
* NIO客户端
*
*/
public class NIOClient {
private Selector selector;
/**
* 获得一个Socket通道，并对该通道做一些初始化的工作
*
* @param ip连接的服务器的ip
* @param port连接的服务器的端口号
* @throws IOException
*/
public void initClient(String ip, int port) throws IOException {
selector = Selector.open();
// 获得一个Socket通道，并设置为非阻塞
SocketChannel channel = SocketChannel.open();
channel.configureBlocking(false);
// 客户端连接服务器,其实方法执行并没有实现连接，需要在listen()方法中调
// 用channel.finishConnect();才能完成连接
channel.connect(new InetSocketAddress(ip, port));
// 将通道管理器和该通道绑定，并为该通道注册SelectionKey.OP_CONNECT事件。
channel.register(selector, SelectionKey.OP_CONNECT);
}
/**
* 循环监听selector上是否有需要处理的事件，如果有，则进行处理
*
* @throws IOException
*/
public void listen() throws IOException {
// 循环访问selector
while (true) {
// //当注册的事件到达时，方法返回；否则,该方法会一直阻塞
selector.select();
// 返回发生了事件的SelectionKey对象的一个集合
Set<SelectionKey> selectedKeys = selector.selectedKeys();
Iterator<SelectionKey> it = selectedKeys.iterator();
while (it.hasNext()) {
SelectionKey key = (SelectionKey) it.next();
it.remove(); // 删除已选的key以防重复处理
// 连接事件发生
if ((key.readyOps() & SelectionKey.OP_CONNECT) == SelectionKey.OP_CONNECT) {
// 获得和服务器连接的通道
SocketChannel channel = (SocketChannel) key.channel();
// 如果正在连接，则完成连接
if (channel.isConnectionPending()) {
channel.finishConnect();
}
channel.configureBlocking(false);// 设置成非阻塞
// 在和服务器连接成功之后，为了可以向服务器发送信息，需要为通道注册写事件。
channel.register(this.selector, SelectionKey.OP_WRITE);
} else if ((key.readyOps() & SelectionKey.OP_WRITE) == SelectionKey.OP_WRITE) {
// 得到可写事件发生的Socket通道
SocketChannel channel = (SocketChannel) key.channel();
String msg = new String("Hello, Nice to meet you!");
channel.write(ByteBuffer.wrap(msg.getBytes())); // 向服务器发送信息
// 在向服务器写完信息之后，为了可以接收到服务器返回的信息，需要给通道注册读事件。
channel.register(this.selector, SelectionKey.OP_READ);
System.out.println("客户端发送信息：" + msg);
} else if ((key.readyOps() & SelectionKey.OP_READ) == SelectionKey.OP_READ) {
// 得到可读事件发生的Socket通道
SocketChannel channel = (SocketChannel) key.channel();
ByteBuffer echoBuffer = ByteBuffer.allocate(1024);
while (true) {
echoBuffer.clear();
int r = channel.read(echoBuffer);
// 判断拷贝是否完成
if (r <= 0) {
break;
}
}
byte[] data = echoBuffer.array();
System.out.println("客户端收到信息：" + new String(data).trim());
}
}
}
}
/**
* 启动客户端测试
*
* @throws IOException
*/
public static void main(String[] args) throws IOException {
NIOClient client = new NIOClient();
client.initClient("localhost", 8000);
client.listen();
}
}

输出结果为：

服务端：

Java NIO (三)

客户端：

Java NIO (三)

相关推荐