摘要: 本系列基于炼数成金课程，为了更好的学习，做了系列的记录。 本文主要介绍

  • 什么是NIO
  • Buffer
  • Channel
  • 网络编程
  • AIO

IO感觉上和多线程并没有多大关系，但是NIO改变了线程在应用层面使用的方式，也解决了一些实际的困难。而AIO是异步IO和前面的系列也有点关系。在此，为了学习和记录，也写一篇文章来介绍NIO和AIO。

1. 什么是NIO

NIO是New I/O的简称，与旧式的基于流的I/O方法相对，从名字看，它表示新的一套Java I/O标 准。它是在Java 1.4中被纳入到JDK中的，并具有以下特性

  • NIO是基于块（Block）的，它以块为基本单位处理数据 （硬盘上存储的单位也是按Block来存储，这样性能上比基于流的方式要好一些）
  • 为所有的原始类型提供（Buffer）缓存支持
  • 增加通道（Channel）对象，作为新的原始 I/O 抽象
  • 支持锁（我们在平时使用时经常能看到会出现一些.lock的文件，这说明有线程正在使用这把锁，当线程释放锁时，会把这个文件删除掉，这样其他线程才能继续拿到这把锁）和内存映射文件的文件访问接口
  • 提供了基于Selector的异步网络I/O

nio和aio

所有的从通道中的读写操作，都要经过Buffer，而通道就是io的抽象，通道的另一端就是操纵的文件。

2. Buffer

nio和aio

Java中Buffer的实现。基本的数据类型都有它对应的Buffer

Buffer的简单使用例子：

  1. package test;
  3. import java.io.File;
  4. import java.io.FileInputStream;
  5. import java.nio.ByteBuffer;
  6. import java.nio.channels.FileChannel;
  8. public class Test {
  9.     public static void main(String[] args) throws Exception {
  10.         FileInputStream fin = new FileInputStream(new File(
  11.                 "d:\\temp_buffer.tmp"));
  12.         FileChannel fc = fin.getChannel();
  13.         ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
  14.         fc.read(byteBuffer);
  15.         fc.close();
  16.         byteBuffer.flip();//读写转换
  17.     }
  18. }
  1. 得到Channel
  2. 申请Buffer
  3. 建立Channel和Buffer的读/写关系
  4. 关闭

下面的例子是使用NIO来复制文件：

  1. public static void nioCopyFile(String resource, String destination)
  2.             throws IOException {
  3.         FileInputStream fis = new FileInputStream(resource);
  4.         FileOutputStream fos = new FileOutputStream(destination);
  5.         FileChannel readChannel = fis.getChannel(); // 读文件通道
  6.         FileChannel writeChannel = fos.getChannel(); // 写文件通道
  7.         ByteBuffer buffer = ByteBuffer.allocate(1024); // 读入数据缓存
  8.         while (true) {
  9.             buffer.clear();
  10.             int len = readChannel.read(buffer); // 读入数据
  11.             if (len == -1) {
  12.                 break; // 读取完毕
  13.             }
  14.             buffer.flip();
  15.             writeChannel.write(buffer); // 写入文件
  16.         }
  17.         readChannel.close();
  18.         writeChannel.close();
  19.     }

nio

这里要区别下容量和上限，比如一个Buffer有10KB，那么10KB就是容量，我将5KB的文件读到Buffer中，那么上限就是5KB。

下面举个例子来理解下这3个重要的参数：

  1. public static void main(String[] args) throws Exception {
  2.         ByteBuffer b = ByteBuffer.allocate(15); // 15个字节大小的缓冲区
  3.         System.out.println("limit=" + b.limit() + " capacity=" + b.capacity()
  4.                 + " position=" + b.position());
  5.         for (int i = 0; i < 10; i++) {
  6.             // 存入10个字节数据
  7.             b.put((byte) i);
  8.         }
  9.         System.out.println("limit=" + b.limit() + " capacity=" + b.capacity()
  10.                 + " position=" + b.position());
  11.         b.flip(); // 重置position
  12.         System.out.println("limit=" + b.limit() + " capacity=" + b.capacity()
  13.                 + " position=" + b.position());
  14.         for (int i = 0; i < 5; i++) {
  15.             System.out.print(b.get());
  16.         }
  17.         System.out.println();
  18.         System.out.println("limit=" + b.limit() + " capacity=" + b.capacity()
  19.                 + " position=" + b.position());
  20.         b.flip();
  21.         System.out.println("limit=" + b.limit() + " capacity=" + b.capacity()
  22.                 + " position=" + b.position());
  24.     }

nio和aio

此时position从0到10，capactiy和limit不变。

nio和aio

该操作会重置position，通常，将buffer从写模式转换为读 模式时需要执行此方法 flip()操作不仅重置了当前的position为0，还将limit设置到当前position的位置 。

limit的意义在于，来确定哪些数据是有意义的，换句话说，从position到limit之间的数据才是有意义的数据，因为是上次操作的数据。所以flip操作往往是读写转换的意思。

nio和aio

意义同上。

而Buffer中大多数的方法都是去改变这3个参数来达到某些功能的：

  1. public final Buffer rewind()
  1. public final Buffer clear()
  1. public final Buffer flip()

2.1 文件映射到内存

  1. public static void main(String[] args) throws Exception {
  2.         RandomAccessFile raf = new RandomAccessFile("C:\\mapfile.txt", "rw");
  3.         FileChannel fc = raf.getChannel();
  4.         // 将文件映射到内存中
  5.         MappedByteBuffer mbb = fc.map(FileChannel.MapMode.READ_WRITE, 0,
  6.                 raf.length());
  7.         while (mbb.hasRemaining()) {
  8.             System.out.print((char) mbb.get());
  9.         }
  10.         mbb.put(0, (byte) 98); // 修改文件
  11.         raf.close();
  12.     }

3. Channel

多线程网络服务器的一般结构：

nio和aio

简单的多线程服务器：

  1. public static void main(String[] args) throws Exception {
  2.         ServerSocket echoServer = null;
  3.         Socket clientSocket = null;
  4.         try {
  5.             echoServer = new ServerSocket(8000);
  6.         } catch (IOException e) {
  7.             System.out.println(e);
  8.         }
  9.         while (true) {
  10.             try {
  11.                 clientSocket = echoServer.accept();
  12.                 System.out.println(clientSocket.getRemoteSocketAddress()
  13.                         + " connect!");
  14.                 tp.execute(new HandleMsg(clientSocket));
  15.             } catch (IOException e) {
  16.                 System.out.println(e);
  17.             }
  18.         }
  19.     }

这里的tp是一个线程池，HandleMsg是处理消息的类。

  1. static class HandleMsg implements Runnable{  
  2.          省略部分信息                 
  3.          public void run(){         
  4.              try {         
  5.                  is = new BufferedReader(new InputStreamReader(clientSocket.getInputStream())); 
  6.                  os = new PrintWriter(clientSocket.getOutputStream(), true); 
  7.                  // 从InputStream当中读取客户端所发送的数据              
  8.                  String inputLine = null;                 
  9.                  long b=System. currentTimeMillis ();                 
  10.                  while ((inputLine = is.readLine()) != null)
  11.                  {           
  12.                      os.println(inputLine);                 
  13.                  }                 
  14.                  long e=System. currentTimeMillis ();                 
  15.                  System. out.println ("spend:"+(e - b)+" ms ");             
  16.             } catch (IOException e) {                 
  17.                 e.printStackTrace();             
  18.             }finally
  19.             {  
  20.                 关闭资源 
  21.             }     
  22.         } 
  23.      }
  1. public static void main(String[] args) throws Exception {
  2.         Socket client = null;
  3.         PrintWriter writer = null;
  4.         BufferedReader reader = null;
  5.         try {
  6.             client = new Socket();
  7.             client.connect(new InetSocketAddress("localhost", 8000));
  8.             writer = new PrintWriter(client.getOutputStream(), true);
  9.             writer.println("Hello!");
  10.             writer.flush();
  11.             reader = new BufferedReader(new InputStreamReader(
  12.                     client.getInputStream()));
  13.             System.out.println("from server: " + reader.readLine());
  14.         } catch (Exception e) {
  15.         } finally {
  16.             // 省略资源关闭
  17.         }
  18.     }

为每一个客户端使用一个线程，如果客户端出现延时等异常，线程可能会被占用很长时间。因为数据的准备和读取都在这个线程中。此时，如果客户端数量众多，可能会消耗大量的系统资源。

解决方案:

使用非阻塞的NIO （读取数据不等待，数据准备好了再工作）

为了体现NIO使用的高效。这里先模拟一个低效的客户端来模拟因网络而延时的情况：

  1. private static ExecutorService tp= Executors.newCachedThreadPool();  
  2.         private static final int sleep_time=1000*1000*1000;  
  3.         public static class EchoClient implements Runnable{   
  4.             public void run(){          
  5.                 try {              
  6.                     client = new Socket();              
  7.                     client.connect(new InetSocketAddress("localhost", 8000)); 
  8.                     writer = new PrintWriter(client.getOutputStream(), true); 
  9.                     writer.print("H");              
  10.                     LockSupport.parkNanos(sleep_time);       
  11.                     writer.print("e");           
  12.                     LockSupport.parkNanos(sleep_time);      
  13.                     writer.print("l");       
  14.                     LockSupport.parkNanos(sleep_time);  
  15.                     writer.print("l");       
  16.                     LockSupport.parkNanos(sleep_time);  
  17.                     writer.print("o");     
  18.                     LockSupport.parkNanos(sleep_time);  
  19.                     writer.print("!");         
  20.                     LockSupport.parkNanos(sleep_time);    
  21.                     writer.println();      
  22.                     writer.flush(); 
  23.                 }catch(Exception e)
  24.                 {
  25.                 }
  26.             }
  27.         }
  1. spend:6000ms 
  2. spend:6000ms 
  3. spend:6000ms 
  4. spend:6001ms 
  5. spend:6002ms 
  6. spend:6002ms 
  7. spend:6002ms 
  8. spend:6002ms 
  9. spend:6003ms 
  10. spend:6003ms
  1. while ((inputLine = is.readLine()) != null)

如果用NIO来处理这个问题会怎么做呢？

NIO有一个很大的特点就是：把数据准备好了再通知我

nio和aio

selector是一个选择器，它可以选择某一个Channel，然后做些事情。

一个线程可以对应一个selector，而一个selector可以轮询多个Channel，而每个Channel对应了一个Socket。

与上面一个线程对应一个Socket相比，使用NIO后，一个线程可以轮询多个Socket。

当selector调用select()时，会查看是否有客户端准备好了数据。当没有数据被准备好时，select()会阻塞。平时都说NIO是非阻塞的，但是如果没有数据被准备好还是会有阻塞现象。

当有数据被准备好时，调用完select()后，会返回一个SelectionKey，SelectionKey表示在某个selector上的某个Channel的数据已经被准备好了。

只有在数据准备好时，这个Channel才会被选择。

这样NIO实现了一个线程来监控多个客户端。

而刚刚模拟的网络延迟的客户端将不会影响NIO下的线程，因为某个Socket网络延迟时，数据还未被准备好，selector是不会选择它的，而会选择其他准备好的客户端。

selectNow()与select()的区别在于，selectNow()是不阻塞的，当没有客户端准备好数据时，selectNow()不会阻塞，将返回0，有客户端准备好数据时，selectNow()返回准备好的客户端的个数。

主要代码：

  1. package test;
  3. import java.net.InetAddress;
  4. import java.net.InetSocketAddress;
  5. import java.net.Socket;
  6. import java.nio.ByteBuffer;
  7. import java.nio.channels.SelectionKey;
  8. import java.nio.channels.Selector;
  9. import java.nio.channels.ServerSocketChannel;
  10. import java.nio.channels.SocketChannel;
  11. import java.nio.channels.spi.AbstractSelector;
  12. import java.nio.channels.spi.SelectorProvider;
  13. import java.util.HashMap;
  14. import java.util.Iterator;
  15. import java.util.LinkedList;
  16. import java.util.Map;
  17. import java.util.Set;
  18. import java.util.concurrent.ExecutorService;
  19. import java.util.concurrent.Executors;
  21. public class MultiThreadNIOEchoServer {
  22.     public static Map<Socket, Long> geym_time_stat = new HashMap<Socket, Long>();
  24.     class EchoClient {
  25.         private LinkedList<ByteBuffer> outq;
  27.         EchoClient() {
  28.             outq = new LinkedList<ByteBuffer>();
  29.         }
  31.         public LinkedList<ByteBuffer> getOutputQueue() {
  32.             return outq;
  33.         }
  35.         public void enqueue(ByteBuffer bb) {
  36.             outq.addFirst(bb);
  37.         }
  38.     }
  40.     class HandleMsg implements Runnable {
  41.         SelectionKey sk;
  42.         ByteBuffer bb;
  44.         public HandleMsg(SelectionKey sk, ByteBuffer bb) {
  45.             super();
  46.             this.sk = sk;
  47.             this.bb = bb;
  48.         }
  50.         @Override
  51.         public void run() {
  52.             // TODO Auto-generated method stub
  53.             EchoClient echoClient = (EchoClient) sk.attachment();
  54.             echoClient.enqueue(bb);
  55.             sk.interestOps(SelectionKey.OP_READ | SelectionKey.OP_WRITE);
  56.             selector.wakeup();
  57.         }
  59.     }
  61.     private Selector selector;
  62.     private ExecutorService tp = Executors.newCachedThreadPool();
  64.     private void startServer() throws Exception {
  65.         selector = SelectorProvider.provider().openSelector();
  66.         ServerSocketChannel ssc = ServerSocketChannel.open();
  67.         ssc.configureBlocking(false);
  68.         InetSocketAddress isa = new InetSocketAddress(8000);
  69.         ssc.socket().bind(isa);
  70.         // 注册感兴趣的事件，此处对accpet事件感兴趣
  71.         SelectionKey acceptKey = ssc.register(selector, SelectionKey.OP_ACCEPT);
  72.         for (;;) {
  73.             selector.select();
  74.             Set readyKeys = selector.selectedKeys();
  75.             Iterator i = readyKeys.iterator();
  76.             long e = 0;
  77.             while (i.hasNext()) {
  78.                 SelectionKey sk = (SelectionKey) i.next();
  79.                 i.remove();
  80.                 if (sk.isAcceptable()) {
  81.                     doAccept(sk);
  82.                 } else if (sk.isValid() && sk.isReadable()) {
  83.                     if (!geym_time_stat.containsKey(((SocketChannel) sk
  84.                             .channel()).socket())) {
  85.                         geym_time_stat.put(
  86.                                 ((SocketChannel) sk.channel()).socket(),
  87.                                 System.currentTimeMillis());
  88.                     }
  89.                     doRead(sk);
  90.                 } else if (sk.isValid() && sk.isWritable()) {
  91.                     doWrite(sk);
  92.                     e = System.currentTimeMillis();
  93.                     long b = geym_time_stat.remove(((SocketChannel) sk
  94.                             .channel()).socket());
  95.                     System.out.println("spend:" + (e - b) + "ms");
  96.                 }
  97.             }
  98.         }
  99.     }
  101.     private void doWrite(SelectionKey sk) {
  102.         // TODO Auto-generated method stub
  103.         SocketChannel channel = (SocketChannel) sk.channel();
  104.         EchoClient echoClient = (EchoClient) sk.attachment();
  105.         LinkedList<ByteBuffer> outq = echoClient.getOutputQueue();
  106.         ByteBuffer bb = outq.getLast();
  107.         try {
  108.             int len = channel.write(bb);
  109.             if (len == -1) {
  110.                 disconnect(sk);
  111.                 return;
  112.             }
  113.             if (bb.remaining() == 0) {
  114.                 outq.removeLast();
  115.             }
  116.         } catch (Exception e) {
  117.             // TODO: handle exception
  118.             disconnect(sk);
  119.         }
  120.         if (outq.size() == 0) {
  121.             sk.interestOps(SelectionKey.OP_READ);
  122.         }
  123.     }
  125.     private void doRead(SelectionKey sk) {
  126.         // TODO Auto-generated method stub
  127.         SocketChannel channel = (SocketChannel) sk.channel();
  128.         ByteBuffer bb = ByteBuffer.allocate(8192);
  129.         int len;
  130.         try {
  131.             len = channel.read(bb);
  132.             if (len < 0) {
  133.                 disconnect(sk);
  134.                 return;
  135.             }
  136.         } catch (Exception e) {
  137.             // TODO: handle exception
  138.             disconnect(sk);
  139.             return;
  140.         }
  141.         bb.flip();
  142.         tp.execute(new HandleMsg(sk, bb));
  143.     }
  145.     private void disconnect(SelectionKey sk) {
  146.         // TODO Auto-generated method stub
  147.         //省略略干关闭操作
  148.     }
  150.     private void doAccept(SelectionKey sk) {
  151.         // TODO Auto-generated method stub
  152.         ServerSocketChannel server = (ServerSocketChannel) sk.channel();
  153.         SocketChannel clientChannel;
  154.         try {
  155.             clientChannel = server.accept();
  156.             clientChannel.configureBlocking(false);
  157.             SelectionKey clientKey = clientChannel.register(selector,
  158.                     SelectionKey.OP_READ);
  159.             EchoClient echoClinet = new EchoClient();
  160.             clientKey.attach(echoClinet);
  161.             InetAddress clientAddress = clientChannel.socket().getInetAddress();
  162.             System.out.println("Accepted connection from "
  163.                     + clientAddress.getHostAddress());
  164.         } catch (Exception e) {
  165.             // TODO: handle exception
  166.         }
  167.     }
  169.     public static void main(String[] args) {
  170.         // TODO Auto-generated method stub
  171.         MultiThreadNIOEchoServer echoServer = new MultiThreadNIOEchoServer();
  172.         try {
  173.             echoServer.startServer();
  174.         } catch (Exception e) {
  175.             // TODO: handle exception
  176.         }
  178.     }
  180. }

当用之前模拟的那个延迟的客户端时，这次的时间消耗就在2ms到11ms之间了。性能提升是很明显的。

总结：

  1. NIO会将数据准备好后，再交由应用进行处理，数据的读取/写入过程依然在应用线程中完成，只是将等待的时间剥离到单独的线程中去。

  2. 节省数据准备时间（因为Selector可以复用）

5. AIO

AIO的特点：

  1. 读完了再通知我

  2. 不会加快IO，只是在读完后进行通知

  3. 使用回调函数，进行业务处理

AIO的相关代码：

AsynchronousServerSocketChannel

  1. server = AsynchronousServerSocketChannel.open().bind( new InetSocketAddress (PORT));
  1. public abstract <A> void accept(A attachment, CompletionHandler<AsynchronousSocketChannel,? super A> handler);

示例代码：

  1. server.accept(null,
  2.                 new CompletionHandler<AsynchronousSocketChannel, Object>() {
  3.                     final ByteBuffer buffer = ByteBuffer.allocate(1024);
  5.                     public void completed(AsynchronousSocketChannel result,
  6.                             Object attachment) {
  7.                         System.out.println(Thread.currentThread().getName());
  8.                         Future<Integer> writeResult = null;
  9.                         try {
  10.                             buffer.clear();
  11.                             result.read(buffer).get(100, TimeUnit.SECONDS);
  12.                             buffer.flip();
  13.                             writeResult = result.write(buffer);
  14.                         } catch (InterruptedException | ExecutionException e) {
  15.                             e.printStackTrace();
  16.                         } catch (TimeoutException e) {
  17.                             e.printStackTrace();
  18.                         } finally {
  19.                             try {
  20.                                 server.accept(null, this);
  21.                                 writeResult.get();
  22.                                 result.close();
  23.                             } catch (Exception e) {
  24.                                 System.out.println(e.toString());
  25.                             }
  26.                         }
  27.                     }
  29.                     @Override
  30.                     public void failed(Throwable exc, Object attachment) {
  31.                         System.out.println("failed: " + exc);
  32.                     }
  33.                 });

这里使用了Future来实现即时返回，关于Future请参考上一篇

在理解了NIO的基础上，看AIO，区别在于AIO是等读写过程完成后再去调用回调函数。

NIO是同步非阻塞的

AIO是异步非阻塞的

由于NIO的读写过程依然在应用线程里完成，所以对于那些读写过程时间长的，NIO就不太适合。

而AIO的读写过程完成后才被通知，所以AIO能够胜任那些重量级，读写过程长的任务。

原文链接：https://my.oschina.net/hosee/blog/615269