为什么一个已经存在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的另一特点是它对于应用程序的表现力的关注,这个我们会在下面的练习中看到。
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,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