JDK的NIO

作者: 没意思先生1995 | 来源:发表于2017-04-15 19:45 被阅读0次

netty极简教程（二）： nio Buffer的原理及使用
JNIO与网络编程
Java NIO（一）
Tomcat NIO2 网络模型原理分
NIO中存在的bug—epoll空轮询
java nio
JDK的NIO
Java-nio
IO流（二）Java的NIO
IO与NIO概览一

文章内容来自李林峰的《Netty权威指南》

JDK的NIO类库

缓冲区Buffer
- ByteBuffer:字节缓冲区
- CharBuffer:字符缓冲区
- ShortBuffer:短整型缓冲区
- IntBuffer:整型缓冲区
- LongBuffer:长整型缓冲区
- FloatBuffer:浮点型缓冲区
- DoubleBuffer:双精度浮点型缓冲区
通道Channel
网络数据通过channel读取和写入，全双工模式
多路复用器Selector
提供选择已就绪任务的能力,selector会不断轮询注册在上面的channel,某个channel发生读或写事件,就处于就绪状态,会被selector轮询出来,然后通过SelectionKey获取就绪Channel的集合，进行后续IO操作

NIO服务端

序列图

NIO服务端序列图

步骤

打开ServerSocketChannel,用于监听客户端的连接，是所有客户端连接的父管道

 ServerSocketChannel serverChannel=ServerSocketChannel.open();

绑定监听端口，设置为非阻塞模式

serverChannel.configureBlocking(false);
serverChannel.socket().bind(new InetSocketAddress(port),1024);
(注:1024指serverChannel的backlog为1024)

创建Reactor线程，创建多路复用器Selector并启动线程

selector=Selector.open();
new Thread(new ReactorTask()).start();

将ServerSocketChannel 注册到Reactor线程的多路复用器Selector,监听accept事件

serverChannel.register(selector,SelectionKey.OP_ACCEPT);

多路复用器在线程run方法的无限循环体内轮询准备就绪的key

selector.select(1000);
Set<SelectionKey> selectedKeys = selector.selectedKeys();
Iterator<SelectionKey> iterator = selectedKeys.iterator();
SelectionKey key = null;
while (iterator.hasNext()){
    key = iterator.next();
    iterator.remove();
    try {
      handleInput(key);
    } catch (Exception e) {
      if(key!=null){
        key.cancel();
        if(key.channel()!=null){
          key.channel().close();
        }
      }
    }
}

多路复用器监听到新客户端的接入,处理新的接入请求,完成TCP三次握手，建立物理链路

SocketChannel sc=serverChannel.accept();

设置客户端链路为非阻塞模式

sc.configureBlocking(false);

将新接入的客户端连接注册到Reactor线程的多路复用器，监听读操作，读取客户端发送的网络消息

sc.register(selector, SelectionKey.OP_READ);

异步读取客户端请求消息到缓冲区

ByteBuffer readBuffer=ByteBuffer.allocate(1024);
int readBytes=sc.read(readBuffer);

对ByteBuffer进行编解码,如果有半包消息指针reset,继续读取后续的报文,将解码成功的消息封装成task,放入线程池中,进行业务操作

  List<Object> messageList=null;
  while(buffer.hasRemain()){
      byteBuffer.mark();
      Object message=decode(byteBuffer);
      if(message==null){
        byteBuffer.reset();
        break;
      }
      messageList.add(message);
  }
  if(!byteBuffer.hasRemain()){
    byteBuffer.clear();
  }else{
     byteBuffer.compact(); 
  }
  if(!messageList!=null & !messageList.isEmpty()){
    for(Object messageE : messageList){
        handlerTask(messageE);
      }
  }

将POJO对象encode成ByteBuffer,调用SocketChannel的异步write接口,将消息异步发送给客户端

channel.write(buffer);

NIO客户端

序列图

NIO客户端序列图
步骤
1. 打开SocketChannel,绑定客户端本地地址
```
SocketChannel socketChannel = SocketChannel.open();
```
1. 设置SocketChannel 为非阻塞模式,同时设置客户端连接TCP参数
```
socketChannel.configureBlocking(false);
socket.setSendBufferSize(BUFFER_SIZE);
......
```
1. 异步连接服务端
```
socketChannel.connect(new InetSocketAddress(host, port);
```
1. 判断是否连接成功,如果连接成功,注册SelectionKey.OP_READ到多路复用器，如果没有说明服务端没返回TCP握手应答，链路没有建立，需要将socketChannel注册到selector,注册SelectionKey.OP_CONNECT
```
if (socketChannel.connect(new InetSocketAddress(host, port))) {
      socketChannel.register(selector, SelectionKey.OP_READ);
} else {
      socketChannel.register(selector, SelectionKey.OP_CONNECT);
 }
```
1. 向Reactor线程的多路复用器Selector注册OP_CONNECT状态位,监听服务端ACK应答

  socketChannel.register(selector, SelectionKey.OP_CONNECT);

创建selector，创建多路复用器并启动线程

  selector = Selector.open();
  new Thread(new ReactorTask()).start();

selector在无限循环体轮询就绪的key

 selector.select(1000);
  Set<SelectionKey> selectedKeys = selector.selectedKeys();
  Iterator<SelectionKey> iterator = selectedKeys.iterator();
  SelectionKey key = null;
  while (iterator.hasNext()){
      key = iterator.next();
      iterator.remove();
      try {
        handleInput(key);
      } catch (Exception e) {
        if(key!=null){
          key.cancel();
          if(key.channel()!=null){
            key.channel().close();
          }
        }
      }
  }

接受connect事件并处理

  if (key.isConnectable()) {
        handlerConnect();
  }

判断连接是否完成，完成则注册读操作到selector

  if (key.isReadable()) {
        registerRead();
  }

向多路复用器selector注册OP_READ事件

  socketChannel.register(selector, SelectionKey.OP_READ);

异步读请求消息到ByteBuffer

  int readBytes = sc.read(byteBuffer);

对ByteBuffer进行编解码,如果有半包消息指针reset,继续读取后续的报文,将解码成功的消息封装成task,放入线程池中,进行业务操作

  List<Object> messageList=null;
  while(buffer.hasRemain()){
      byteBuffer.mark();
      Object message=decode(byteBuffer);
      if(message==null){
        byteBuffer.reset();
        break;
      }
      messageList.add(message);
  }
  if(!byteBuffer.hasRemain()){
    byteBuffer.clear();
  }else{
     byteBuffer.compact(); 
  }
  if(!messageList!=null & !messageList.isEmpty()){
    for(Object messageE : messageList){
        handlerTask(messageE);
      }
  }