java NIO 的最佳实践

NIO的背景

为什么一个已经存在10年的增强包还是JAVA的新I/O包呢？原因是对于大多数的Java程序员而言，基本的I/O操作都能够胜任。在日常工作中，大部分的Java开发者没有必要去学习NIO。更进一步，NIO不仅仅是一个性能提升包。相反，它是一个和Java I/O相关的不同功能的集合。NIO通过使得Java应用的性能“更加接近实质”来达到性能提升的效果，也就是意味着NIO和NIO.2的API暴露了低层次的系统操作的入口。NIO的代价就是它在提供更强大的I/O控制能力的同时，也要求我们比使用基本的I/O编程更加细心地使用和练习。NIO的另一特点是它对于应用程序的表现力的关注，这个我们会在下面的练习中看到。

Java NIO和IO的主要区别

1. 面向流与面向缓冲. Java NIO和IO之间第一个最大的区别是，IO是面向流的，NIO是面向缓冲区的。Java IO面向流意味着每次从流中读一个或多个字节，直至读取所有字节，它们没有被缓存在任何地方。此外，它不能前后移动流中的数据。如果需要前后移动从流中读取的数据，需要先将它缓存到一个缓冲区。 Java NIO的缓冲导向方法略有不同。数据读取到一个它稍后处理的缓冲区，需要时可在缓冲区中前后移动。这就增加了处理过程中的灵活性。
2. 阻塞与非阻塞IO Java IO的各种流是阻塞的。这意味着，当一个线程调用read() 或 write()时，该线程被阻塞，直到有一些数据被读取，或数据完全写入。该线程在此期间不能再干任何事情了。 Java NIO的非阻塞模式，使一个线程从某通道发送请求读取数据，但是它仅能得到目前可用的数据，如果目前没有数据可用时，该线程可以继续做其他的事情。 非阻塞写也是如此。一个线程请求写入一些数据到某通道，但不需要等待它完全写入，这个线程同时可以去做别的事情。线程通常将非阻塞IO的空闲时间用于在其它通道上执行IO操作，所以一个单独的线程现在可以管理多个输入和输出通道（channel）。
3. 选择器（Selectors） Java NIO的选择器允许一个单独的线程来监视多个输入通道，你可以注册多个通道使用一个选择器，然后使用一个单独的线程来“选择”通道：这些通道里已经有可以处理的输入，或者选择已准备写入的通道。这种选择机制，使得一个单独的线程很容易来管理多个通道。

最佳practice

SelectionKey.OP_WRITE订阅时机

现象: cpu占用超高

原因: 订阅了SelectionKey.OP_WRITE事件

Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
 while (iterator.hasNext()) {
 SelectionKey selectionKey = iterator.next();
 iterator.remove();
 if (selectionKey.isConnectable()) {
 SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
 if (socketChannel.isConnectionPending()) {
 socketChannel.finishConnect();
 }
 socketChannel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE);
 }

分析: 当socket缓冲区可写入时就会触发OP_WRITE事件. 而socket缓冲区大多时间都可写入(网络不拥堵),由于nio水平触发的特性OP_WRITE会一直触发导致while()一直空转

水平触发: 简单解释为只要满足条件就一直触发,而不是发生状态改变时才触发(有点主动和被动触发的感觉)

最佳实践:

方案一: 当有写数据需求时订阅OP_WRITE事件,数据发送完成取消订阅.

	while (channel.isOpen()) {
 if (channel.isConnected() && writeBuffer.isReadable()) {
 //writeBuffer可读 注册write事件
 channel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE);
 }
 		//当采用临时订阅OP_WRITE方式 必须使用select(ms)进行超时返回
 // 因为很有可能当select()前极短时间内writeBuffer有数据,而此时没有订阅OP_WRITE事件,会使select()一直阻塞
 		int ready = selector.select(300);
 if (ready > 0) {
 	SelectionKey selectionKey = iterator.next();
 iterator.remove();
 SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
 socketChannel.configureBlocking(false);
 if (selectionKey.isWritable()) {
 	 writeBuffer.flip();
 while (writeBuffer.hasRemaining()) {
 channel.write(writeBuffer);
 }
 writeBuffer.clear();
 	 socketChannel.register(selector, SelectionKey.OP_READ);
 }
 }
	}

当使用临时订阅OP_WRITE事件方式时,必须使用selector.select(long),进行超时返回. 因为很有可能当select()前极短时间内writeBuffer有数据,而此时没有订阅OP_WRITE事件,会使select()一直阻塞

方案二: 不订阅OP_WRITE事件,直接通过socketChannel.write()写数据.

 	 Selector selector = Selector.open();
 channel.register(selector, SelectionKey.OP_CONNECT);
 channel.connect(new InetSocketAddress("localhost", 5555));
 while (channel.isOpen()) {
 if (channel.isConnected()) {
 writeBuffer.flip();
 while (writeBuffer.hasRemaining()) {
 channel.write(writeBuffer);
 }
 writeBuffer.clear();
 }
 int ready = selector.select(500);
 ...各种事件处理
 }

方案三: 一直订阅OP_WRITE,socketChannel主动写

 while (channel.isOpen()) {
 //这里与方案一有区别 可以直接阻塞
 int ready = selector.select();
 if (ready > 0) {
 Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
 while (iterator.hasNext()) {
 ...缓冲区已写数据清理
 
 SelectionKey selectionKey = iterator.next();
 iterator.remove();
 SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
 socketChannel.configureBlocking(false);
 if (selectionKey.isConnectable()) {
 if (socketChannel.isConnectionPending()) {
 socketChannel.finishConnect();
 }
 //订阅读/写事件
 socketChannel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE);
 }
 if (selectionKey.isReadable()) {
 ...读事件处理
 }
 if (selectionKey.isWritable()) {
 //改为主动读取式
 ByteBuffer byteBuffer = awaitGetWrite(writeBuffer, 30, 50);
 if (byteBuffer != null) {
 int write = channel.write(byteBuffer);
 writeBuffer.readerIndex(writeBuffer.readerIndex() + write);
 if (write != byteBuffer.limit()) {
 System.out.print("a");
 }
 }
 }
 }
 }
 }
 
 /**
 * 等待获取写缓存
 * @param byteBuf 缓冲区
 * @param ms 缓冲时间 防止空转
 * @param cap 阈值:超过则直接返回,没超过等待ms后判断是否超过阈值
 * @return
 */
 public ByteBuffer awaitGetWrite(ByteBuf byteBuf, long ms, int cap) {
 //缓冲大小 不要过大就行 自己调整
 int socketCap = 1024 * 30;
 if (byteBuf.readableBytes() >= cap) {//>=cap直接返回
 return ByteBuffer.allocate(byteBuf.readableBytes() > socketCap ? socketCap : byteBuf.readableBytes());
 } else {//<cap时等待
 CountDownLatch countDownLatch = new CountDownLatch(1);
 try {
 countDownLatch.await(ms, TimeUnit.MILLISECONDS);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 if (byteBuf.readableBytes() > 0) {
 return ByteBuffer.allocate(byteBuf.readableBytes() > socketCap ? socketCap : byteBuf.readableBytes());
 } else {
 return null;
 }
 }
 }

优点缺点方案1当网络拥堵时,不尝试写数据需要自己控制订阅/取消订阅的时机方案2不关心网络拥堵,只要有数据就尝试写,当网络拥堵时做大量无用功编写方便,无需关心OP_WRITE事件订阅时机方案3相比方案1 编码复杂度下降

综合上述个人觉得还是方案3比较好

channel.write()写数据问题

现象: 网络拥堵时,cpu占用超高

原因: 网络拥堵时, channel.write()一直写不进去,导致while()空转

采取上一问题方案3可以避免该问题

			 writeBuffer.flip();
 while (writeBuffer.hasRemaining()) {
 channel.write(writeBuffer);
 }
 writeBuffer.clear();

分析: 当网络拥堵时,channel.write()可能写入0数据,而这里采用死循环写入数据,假如一直写不进去就会导致空转

最佳实践:

 	while (writeBuffer.isReadable()) {
 	 //这里使用的是netty的ByteBuf
 ByteBuffer byteBuffer = writeBuffer.nioBuffer();
 channel.write(byteBuffer);
 writeBuffer.readerIndex(writeBuffer.readerIndex() + byteBuffer.position());
 int left = byteBuffer.limit() - byteBuffer.position();
 if (left != 0) {//无法全部写入到socket缓冲区中,说明socket缓冲区已满,可能发生空转 break
 System.err.print("a");
 //防止空转 依赖外层循环重新进入 
 break;
 } 
 }
 

结合OP_WRITE订阅时机问题,可以得知方案一的临时订阅OP_WRITE事件方式,能更好的防止channel.write(byteBuffer)空转

TCP断开判断

现象: 当TCP一方断开时,另一方cpu占用超高

原因: 当TCP一方断开时,一直会触发OP_READ,导致空转.

分析: 当TCP一方断开时,触发OP_READ,socketChannel.read(readBuffer)返回-1,表示对方连接已断开,自己也需要断开连接socketChannel.close(),否则会一直触发OP_READ,导致空转

	while (true) {
 int ready = selector.select();
 if (ready > 0) {
 Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
 while (iterator.hasNext()) {
 SelectionKey selectionKey = iterator.next();
 iterator.remove();
 if (selectionKey.isConnectable()) {
 SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
 if (socketChannel.isConnectionPending()) {
 socketChannel.finishConnect();
 }
 socketChannel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE);
 } else if (selectionKey.isReadable()) {
 SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
 socketChannel.configureBlocking(false);
 	//The number of bytes read, possibly zero, or -1 if the channel has reached end-of-stream
 int read = socketChannel.read(readBuffer);
 readBuffer.flip();
 	//读到-1 没有处理 导致空转
 if (read > 0) {
 System.out.print(new String(readBuffer.array(), 0, read));
 }
 } 
 ...
 }
 }
 }
复制代码

最佳实践:

	if (selectionKey.isReadable()) {
 ByteBuffer readBuffer = Server.SocketContext.get(socketChannel).getReadBuffer();
 int read = socketChannel.read(readBuffer);
 readBuffer.flip();
 if (read > 0) {
 System.out.print(new String(readBuffer.array(), 0, read));
 } else if (read == -1) {//对面已断开 close
 System.out.println("断开..."
 + socketChannel.socket().getRemoteSocketAddress());
 socketChannel.close();
 }
 }

ByteBuf使用

ByteBuf,ByteBuffer对比

特性ByteBuffer1.有position,limit属性,通过flip()切换读写模式 ,不支持同时读/写 2.定长 3.直接内存ByteBuf1.有rix,wix,cap,maxCap属性,支持同时读/写 2.自动扩容 3.直接内存,堆内存,组合

建议使用ByteBuf

ByteBuf 的clear()和discardReadBytes()对比

现象: 使用clear()导致丢数据

原因: clear()实现通过 rix=wix=0,假如此时同时有数据写入,该部分数据则丢失

	if (selectionKey.isWritable()) {
	 while (writeBuffer.isReadable()) {
	 ByteBuffer byteBuffer = writeBuffer.nioBuffer();
	 channel.write(byteBuffer);
	 writeBuffer.readerIndex(writeBuffer.readerIndex() + byteBuffer.position());
	 int left = byteBuffer.limit() - byteBuffer.position();
	 if (left != 0) {//无法一次性写入到缓冲区中,可能发生空转 break
	 		...
	 break;
	 } else {
 //清理已发送数据
	 writeBuffer.clear();
	 }
	 }
	 ...
	}
复制代码

最佳实践:

使用discardReadBytes(),其通过arrayCopy方式并且线程安全,能够防止数据丢失.但频繁的arrayCopy会有性能问题. 可以使用clear()和discardReadBytes()的组合

	if (selectionKey.isWritable()) {
	 while (writeBuffer.isReadable()) {
 	//当缓冲区使用>2/3事 且wix-rix< (maxCap*1/3) 对缓冲区进行整理
 	if (writeBuffer.writerIndex() > (writeBuffer.maxCapacity() / 3 * 2) && writeBuffer.writerIndex() - writeBuffer
 .readerIndex() < (writeBuffer.maxCapacity() / 3)) {
 System.out.println(String.format("缓冲区使用超过2/3 discardReadBytes writerIndex:%d " +
 "readerIndex:%d", writeBuffer
 .writerIndex(), writeBuffer.readerIndex()));
 writeBuffer.discardReadBytes();
 }
 	
	 ByteBuffer byteBuffer = writeBuffer.nioBuffer();
	 channel.write(byteBuffer);
	 writeBuffer.readerIndex(writeBuffer.readerIndex() + byteBuffer.position());
	 int left = byteBuffer.limit() - byteBuffer.position();
	 if (left != 0) {//无法一次性写入到缓冲区中,可能发生空转 break
	 ...
	 //防止空转 等待下次write事件
	 break;
	 } else {
	 //注意clear()的使用 因为writeBuffer一直在写入 writerIndex可能>readIndex
	 if (writeBuffer.writerIndex() == writeBuffer.readerIndex()) {
	 //TODO 因为不是原子过程 理论上会有问题 但实际验证中却没问题 待验证
	 writeBuffer.clear();
	 System.out.println("clear");
	 } 
	 }
	 }
	 ...
	}

使用快速收敛

在GunNetty中，快速收敛确保Selector中所有的key均为有效key，不包含失效key，该方法一般使用在关闭channel之后

@Override
public int fastLimit() throws IOException {
 bootSelector.wakeup();
 return bootSelector.select(0);
}

1.如果正在阻塞轮训，立刻终止，使用wakeup函数

2.立刻select(0)删除已经失效的key