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

JDK的NIO类库

• 缓冲区Buffer

  • ByteBuffer:字节缓冲区
  • CharBuffer:字符缓冲区
  • ShortBuffer:短整型缓冲区
  • IntBuffer:整型缓冲区
  • LongBuffer:长整型缓冲区
  • FloatBuffer:浮点型缓冲区
  • DoubleBuffer:双精度浮点型缓冲区

• 通道Channel
  网络数据通过channel读取和写入，全双工模式

• 多路复用器Selector
  提供选择已就绪任务的能力,selector会不断轮询注册在上面的channel,某个channel发生读或写事件,就处于就绪状态,会被selector轮询出来,然后通过SelectionKey获取就绪Channel的集合，进行后续IO操作

NIO服务端

• 序列图

  
  
  NIO服务端序列图

• 步骤

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

   ServerSocketChannel serverChannel=ServerSocketChannel.open();
  

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

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

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

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

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

  serverChannel.register(selector,SelectionKey.OP_ACCEPT);
  

  5. 多路复用器在线程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();
          }
        }
      }
  } 
  

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

  SocketChannel sc=serverChannel.accept();
  

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

  sc.configureBlocking(false);
  

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

  sc.register(selector, SelectionKey.OP_READ);  
  

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

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

  10. 对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);
        }
    }
  

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

  channel.write(buffer);
  

NIO客户端

• 序列图

  
  
  NIO客户端序列图

• 步骤

  1. 打开SocketChannel,绑定客户端本地地址

  SocketChannel socketChannel = SocketChannel.open();
  

  2. 设置SocketChannel 为非阻塞模式,同时设置客户端连接TCP参数

  socketChannel.configureBlocking(false);
  socket.setSendBufferSize(BUFFER_SIZE);
  ......
  

  3. 异步连接服务端

  socketChannel.connect(new InetSocketAddress(host, port);
  

  4. 判断是否连接成功,如果连接成功,注册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);
   }
  

  5. 向Reactor线程的多路复用器Selector注册OP_CONNECT状态位,监听服务端ACK应答

  socketChannel.register(selector, SelectionKey.OP_CONNECT);

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

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

7. 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();
          }
        }
      }
  } 

8. 接受connect事件并处理

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

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

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

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

  socketChannel.register(selector, SelectionKey.OP_READ);

11. 异步读请求消息到ByteBuffer

  int readBytes = sc.read(byteBuffer);

12. 对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);
      }
  }

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

channel.write(buffer);