java NIO

Posted 2020-07-13 _oldzhang

javaNIO是非阻塞的IO。可以用于替代IO操作，但用于对文件的操作时它并不能设置为非阻塞，它的优势体现在网络通信上。从上一篇文章java网络-Socket来看,即使使用多线程来处理Socket，但一个线程只能处理一个客户端的请求，单个线程在read的时候还是会阻塞，开销还是很大。如果使用NIO来处理，当线程从通道读取数据到缓冲区时，线程还是可以进行其他事情。当数据被写入到缓冲区时，线程可以继续处理它。从缓冲区写入通道也类似。NIO引入了选择器的概念，选择器用于监听多个通道的事件（比如：连接打开，数据到达）。因此，单个的线程可以监听多个数据通道。

标准的IO基于字节流和字符流进行操作的，而NIO是基于通道（Channel）和缓冲区（Buffer）进行操作，数据总是从通道读取到缓冲区中，或者从缓冲区写入到通道中。Channels、Buffers、Selectors是NIO的核心组成部分。

Channel:

Channel类似于IO中的流。主要实现有：FileChannel（用于处理文件）DatagramChannel（处理UDP）SocketChannel(处理TCP连接)、ServerSocketChannel(可以监听TCP连接，和ServerSocket能创建一样它可以创建SocketChannel)，但流是单向读写的，要实现对流的读写需要分别使用input/ouput流，Channel是双向的。而且通道可以异步读写。数据在通道中需要先写入到Buffer中，也只能从Buffer中读取。

Selector：
Selector允许单线程处理多个 Channel。如果你的应用打开了多个连接（通道），但每个连接的流量都很低，使用Selector就会很方便。要使用Selector，得向Selector注册Channel，然后调用它的select()方法。这个方法会一直阻塞到某个注册的通道有事件就绪。一旦这个方法返回，线程就可以处理这些事件，事件的例子有如新连接进来，数据接收等。

File读取文件的一个例子

public class FileChannelTest {
    public static void main(String args[]) throws IOException {
        RandomAccessFile aFile = new RandomAccessFile("nio-data.txt", "rw");
        FileChannel inChannel = aFile.getChannel();
        //定义一个固定大小的buffer
        ByteBuffer buf = ByteBuffer.allocate(48);
        //将数据从channel写入buffer 返回channel字节数
        int bytesRead = inChannel.read(buf);
        while (bytesRead != -1) {
            System.out.println("Read " + bytesRead);
            buf.flip();
            while(buf.hasRemaining()){
                System.out.print((char) buf.get());
            }
            //将buffer清空 并将空的channel写入buf 控制循环结束
            buf.clear();
            bytesRead = inChannel.read(buf);
        }
        aFile.close();
    }
}

Buffer：

缓冲区本质上是一块可以写入数据，然后可以从中读取数据的内存。这块内存被包装成NIO Buffer对象，并提供了一组方法，用来方便的访问该块内存Buffer读写数据一般经过下面四个步骤：
1，写入数据到Buffer
2，调用flip()方法：将Buffer从写模式切换到读模式。在读模式下，可以读取之前写入到buffer的所有数据。
3，从Buffer中读取数据
4，调用clear()方法或者compact()方法：

一旦读完了所有的数据，就需要清空缓冲区，让它可以再次被写入。有两种方式能清空缓冲区：调用clear()或compact()方法。

clear()方法会清空整个缓冲区。

compact()方法只会清除已经读过的数据。任何未读的数据都被移到缓冲区的起始处，新写入的数据将放到缓冲区未读数据的后面。
Buffer有三个重要的属性：capacity、position、limit

capacity指Buffer创建时的容量大小，position和limit取决于当前是读还是写模式。

在写模式下：

position初始为0，当写入一个数据到Buffer后， position会向前移动到下一个可插入数据的Buffer单元。position最大可为capacity – 1。

limit表示可写的最大位置，和capacity相同。

在读模式下：

position会在切换到读模式时重置为0，表示可从最开始位置读取，当读取完一个buffer单元后会后移一个单元到下一次读取的位置。

limit为所有数据占据的最大buffer单元处，也就是在切换前写模式下的position位置。

向Buffer中写数据：

1，从Channel写到Buffer：inChannel.read(buf);

2，使用Buffer的put()方法：buf.put(123);

从Buffer中读取数据：

1，从Buffer读取数据到Channel：inChannel.write(buf);

2，使用get()方法从Buffer中读取数据：byte aByte = buf.get();

Buffer常用方法：

1，flip()方法：将Buffer从写模式切换到读模式，将position设置为0，写模式下的position设置为limit

2，rewind()方法：将position设回为0，所以可以重读Buffer中的所有数据

3，clear()、compact()方法：上面已介绍，实际上clear也不会真正清除数据，只是将position设置为0,可以从0开始写了。如果存在未读数据使用compact()可将未读数据置前。

4，mark()、reset()方法：可以标记Buffer中的一个特定position。之后可以通过调用Buffer.reset()方法恢复到这个position。

通道之间的数据传输：

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

FileChannel的transferFrom()方法可以将数据从源通道传输到FileChannel中。

transferTo()方法将数据从FileChannel传输到其他的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();
toChannel.transferFrom(position, count, fromChannel);
//fromChannel.transferTo(position, count, toChannel);

方法的输入参数position表示从position处开始向目标文件写入数据，count表示最多传输的字节数。

Selector：
Selector（选择器）是Java NIO中能够检测一到多个NIO通道，并能够知晓通道是否为诸如读写事件做好准备的组件。单个线程处理多个Channels的好处是可以减少多线程切换的开销。
Selector的创建
Selector selector = Selector.open();
向Selector注册通道
SelectableChannel.register()方法来实现，如下：
channel.configureBlocking(false);
SelectionKey key = channel.register(selector,Selectionkey.OP_READ);
与Selector一起使用时，Channel必须处于非阻塞模式下。这意味着不能将FileChannel与Selector一起使用，因为FileChannel不能切换到非阻塞模式。而套接字通道都可以。
Selectionkey.OP_READ是一个interest集合，selector监听的事件类型，有Connect/Accept/Read/Write等类型。可以同时监听多个。返回的类型也是SelectionKey类型，包含了interest集合、ready集合、Channel、Selector

interest集合：
注册时监听的事件类型
ready集合：
ready 集合是通道已经准备就绪的操作的集合，可以通过下面四个方法来检测目前就绪的事件是什么事件，从而进行相应的处理
selectionKey.isAcceptable();
selectionKey.isConnectable();
selectionKey.isReadable();
selectionKey.isWritable();
获取Channel和Selector：
Channel  channel  = selectionKey.channel();
Selector selector = selectionKey.selector();
SelectionKey还可以携带附加的对象。

//获取selector对象
Selector selector = Selector.open();
//设置Channel为非阻塞
channel.configureBlocking(false);
//向channel注册选择器 非监听read事件的就绪状态
SelectionKey key = channel.register(selector, SelectionKey.OP_READ);
while(true) {
  //阻塞方法  有read事件就绪时会返回就绪channel的个数
  int readyChannels = selector.select();
  if(readyChannels == 0) continue;
  //访问“已选择键集”中的就绪通道SelectionKey对象 
  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
    }
    //Selector不会自己从已选择键集中移除SelectionKey实例。必须在处理完通道时自己移除。下次该通道变成就绪时，Selector会再次将其放入已选择键集中。
    keyIterator.remove();
  }
}

下面是一个完整的例子，Selector监听Accept事件，监听到新进来的连接创建SocketChannel后又让原来的Selector监听这个通道上的read事件，当read就绪后打印客户端传输过来的数据。

public class Nioserver {
	// 通道管理器
	private Selector selector;
	/**
	 * 获得一个ServerSocket通道，并对该通道做一些初始化的工作
	 * @param port 绑定的端口号
	 * @throws IOException
	 */
	public void initServer(int port) throws IOException {
		// 获得一个ServerSocket通道
		ServerSocketChannel serverChannel = ServerSocketChannel.open();
		// 设置通道为非阻塞
		serverChannel.configureBlocking(false);
		// 将该通道对应的ServerSocket绑定到port端口
		serverChannel.socket().bind(new InetSocketAddress(port));
		// 获得一个通道管理器
		this.selector = Selector.open();
		// 将通道管理器和该通道绑定，并为该通道注册SelectionKey.OP_ACCEPT事件,注册该事件后，
		// 当该事件到达时，selector.select()会返回，如果该事件没到达selector.select()会一直阻塞。
		serverChannel.register(selector, SelectionKey.OP_ACCEPT);
	}
	/**
	 * 采用轮询的方式监听selector上是否有需要处理的事件，如果有，则进行处理
	 * @throws IOException
	 */
	public void listen() throws IOException {
		System.out.println("服务端启动成功！");
		// 轮询访问selector
		while (true) {
			// 当注册的事件到达时，方法返回；否则,该方法会一直阻塞
			selector.select();
			// 获得selector中选中的项的迭代器，选中的项为注册的事件
			Iterator<?> ite = this.selector.selectedKeys().iterator();
			while (ite.hasNext()) {
				SelectionKey key = (SelectionKey) ite.next();
				// 删除已选的key,以防重复处理
				ite.remove();
				handler(key);
			}
		}
	}
	/**
	 * 处理请求
	 * @param key
	 * @throws IOException
	 */
	public void handler(SelectionKey key) throws IOException {
		// 客户端请求连接事件
		if (key.isAcceptable()) {
			handlerAccept(key);
			// 获得了可读的事件
		} else if (key.isReadable()) {
			handelerRead(key);
		}
	}
	/**
	 * 处理连接请求
	 * @param key
	 * @throws IOException
	 */
	public void handlerAccept(SelectionKey key) throws IOException {
		ServerSocketChannel server = (ServerSocketChannel) key.channel();
		// 获得和客户端连接的通道
		SocketChannel channel = server.accept();
		// 设置成非阻塞
		channel.configureBlocking(false);
		// 在这里可以给客户端发送信息哦
		System.out.println("新的客户端连接");
		// 在和客户端连接成功之后，为了可以接收到客户端的信息，需要给通道设置读的权限。
		channel.register(this.selector, SelectionKey.OP_READ);
	}
	/**
	 * 处理读的事件
	 * @param key
	 * @throws IOException
	 */
	public void handelerRead(SelectionKey key) throws IOException {
		// 服务器可读取消息:得到事件发生的Socket通道
		SocketChannel channel = (SocketChannel) key.channel();
		// 创建读取的缓冲区
		ByteBuffer buffer = ByteBuffer.allocate(1024);
		int read = channel.read(buffer);
		if(read > 0){
			byte[] data = buffer.array();
			String msg = new String(data).trim();
			System.out.println("服务端收到信息：" + msg);
			//回写数据
			ByteBuffer outBuffer = ByteBuffer.wrap("好的".getBytes());
			channel.write(outBuffer);// 将消息回送给客户端
		}else{
			System.out.println("客户端关闭");
			key.cancel();
		}
	}
	/**
	 * 启动服务端测试
	 * @throws IOException
	 */
	public static void main(String[] args) throws IOException {
		NIOServer server = new NIOServer();
		server.initServer(8000);
		server.listen();
	}
}