Question

Java NIO是New IO的简称，它是一种可以替代Java IO的一套新的IO机制。它提供了一套不同于Java标准IO的操作机制。严格来说，NIO与并发并无直接的关系。但是，使用NIO技术可以大大提高线程的使用效率。

Java NIO中涉及的基础内容有通道（Channel）和缓冲区（Buffer）、文件IO和网络IO。有关通道、缓冲区以及文件IO在这里不打算进行详细的介绍，大家可以参考本章的参考文献。在这里，我想多花一点时间详细介绍一下有关网络IO的内容。

对于标准的网络IO来说，我们会使用Socket进行网络的读写。为了让服务器可以支持更多的客户端连接，通常的做法是为每一个客户端连接开启一个线程。让我们先回顾一下这方面的内容。

5.10.1　基于Socket的服务端的多线程模式

这里，我以一个简单的Echo服务器为例。对于Echo服务器，它会读取客户端的一个输入，并将这个输入原封不动地返回给客户端。这看起来很简单，但是麻雀虽小五脏俱全。为了完成这个功能，服务器还是需要有一套完整的Socket处理机制。因此，这个Echo服务器非常适合来进行学习。实际上，我认为任何业务逻辑简单的系统都很适合学习，大家不用为了去理解业务上复杂的功能而忽略了系统的重点。

服务端使用多线程进行处理时的结构示意图，如图5.19所示。

图5.19　多线程的服务端

服务器会为每一个客户端连接启用一个线程，这个新的线程将全心全意为这个客户端服务。同时，为了接受客户端连接，服务器还会额外使用一个派发线程。

下面的代码实现了这个服务器：

01 public class MultiThreadEchoServer {
02   private static ExecutorService  tp=Executors.newCachedThreadPool();
03   static class HandleMsg implements Runnable{
04     Socket clientSocket;
05     public HandleMsg(Socket clientSocket){
06       this.clientSocket=clientSocket;
07     }
08
09     public void run(){
10       BufferedReader is =null;
11       PrintWriter os = null;
12       try {
13
14     is = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));
15         os = new PrintWriter(clientSocket.getOutputStream(), true);
16         // 从InputStream当中读取客户端所发送的数据
17         String inputLine = null;
18         long b=System.currentTimeMillis();
19         while ((inputLine = is.readLine()) != null) {
20           os.println(inputLine);
21         }
22         long e=System.currentTimeMillis();
23         System.out.println("spend:"+(e-b)+"ms");
24       } catch (IOException e) {
25         e.printStackTrace();
26       }finally{
27         try {
28           if(is!=null)is.close();
29           if(os!=null)os.close();
30           clientSocket.close();
31         } catch (IOException e) {
32           e.printStackTrace();
33         }
34       }
35     }
36   }
37   public static void main(String args[]) {
38     ServerSocket echoServer = null;
39     Socket clientSocket = null;
40     try {
41       echoServer = new ServerSocket(8000);
42     } catch (IOException e) {
43       System.out.println(e);
44     }
45     while (true) {
46       try {
47         clientSocket = echoServer.accept();
48         System.out.println(clientSocket.getRemoteSocketAddress() + " connect!");
49         tp.execute(new HandleMsg(clientSocket));
50       } catch (IOException e) {
51         System.out.println(e);
52       }
53     }
54   }
55 }

第2行，我们使用了一个线程池来处理每一个客户端连接。第3～33行，定义了HandleMsg线程，它由一个客户端Socket构造而成，它的任务是读取这个Socket的内容并将其进行返回，返回成功后，任务完成，客户端Soceket就被正常关闭。其中第23行，统计并输出了服务端线程处理一次客户端请求所花费的时间（包括读取数据和回写数据的时间）。主线程main的主要作用是在8000端口上进行等待。一旦有新的客户端连接，它就根据这个连接创建HandleMsg线程进行处理（第47～49行）。

这就是一个支持多线程的服务端的核心内容。它的特点是，在相同可支持的线程范围内，可以尽量多地支持客户端的数量，同时和单线程服务器相比，它也可以更好地使用多核CPU。

为了方便大家学习，这里再给出一个客户端的参考实现：

01 public static void main(String[] args) throws IOException {
02   Socket client = null;
03   PrintWriter writer = null;
04   BufferedReader reader = null;
05   try {
06     client = new Socket();
07     client.connect(new InetSocketAddress("localhost", 8000));
08     writer = new PrintWriter(client.getOutputStream(), true);
09     writer.println("Hello!");
10     writer.flush();
11
12     reader = new BufferedReader(new InputStreamReader(client.getInputStream()));
13     System.out.println("from server: " + reader.readLine());
14   } catch (UnknownHostException e) {
15     e.printStackTrace();
16   } catch (IOException e) {
17     e.printStackTrace();
18   } finally {
19     if (writer != null)
20       writer.close();
21     if (reader != null)
22       reader.close();
23     if (client != null)
24       client.close();
25   }
26 }

上述代码在第7行，连接了服务器的8000端口，并发送字符串。接着在第12行，读取服务器的返回信息并进行输出。

可以说，这种多线程的服务器开发模式是极其常用的。对于绝大多数应用来说，这种模式可以很好地工作。但是，如果你想让你的程序工作得更加有效，就必须知道这种模式的一个重大弱点——那就是它倾向于让CPU进行IO等待。为了理解这一点，让我们看一下下面这个比较极端的例子：

01 public class HeavySocketClient {
02   private static ExecutorService  tp=Executors.newCachedThreadPool();
03   private static final int sleep_time=1000*1000*1000;
04   public static class EchoClient implements Runnable{
05     public void run(){
06       Socket client = null;
07       PrintWriter writer = null;
08       BufferedReader reader = null;
09       try {
10         client = new Socket();
11         client.connect(new InetSocketAddress("localhost", 8000));
12         writer = new PrintWriter(client.getOutputStream(), true);
13         writer.print("H");
14         LockSupport.parkNanos(sleep_time);
15         writer.print("e");
16         LockSupport.parkNanos(sleep_time);
17         writer.print("l");
18         LockSupport.parkNanos(sleep_time);
19         writer.print("l");
20         LockSupport.parkNanos(sleep_time);
21         writer.print("o");
22         LockSupport.parkNanos(sleep_time);
23         writer.print("!");
24         LockSupport.parkNanos(sleep_time);
25         writer.println();
26         writer.flush();
27
28         reader = new BufferedReader(new InputStreamReader(client.getInputStream()));
29         System.out.println("from server: " + reader.readLine());
30       } catch (UnknownHostException e) {
31         e.printStackTrace();
32       } catch (IOException e) {
33         e.printStackTrace();
34       } finally {
35         try {
36           if (writer != null)
37             writer.close();
38           if (reader != null)
39             reader.close();
40           if (client != null)
41             client.close();
42         } catch (IOException e) {
43         }
44       }
45     }
46   }
47   public static void main(String[] args) throws IOException {
48     EchoClient ec=new EchoClient();
49     for(int i=0;i＜10;i++)
50       tp.execute(ec);
51   }
52 }

上述代码定义了一个新的客户端，它会进行10次请求（第49～50行开启10个线程）。每一次请求都会访问8000端口。连接成功后，会向服务器输出“Hello!”字符串（第13～26行），但是在这一次交互中，客户端会慢慢地进行输出，每次只输出一个字符，之后进行1秒的等待。因此，整个过程会持续6秒。

开启多线程池的服务器和上述客户端。服务器端的部分输出如下：

spend:6000ms
spend:6000ms
spend:6000ms
spend:6001ms
spend:6002ms
spend:6002ms
spend:6002ms
spend:6002ms
spend:6003ms
spend:6003ms

可以看到，对于服务端来说，每一个请求的处理时间都在6秒左右。这很容易理解，因为服务器要先读入客户端的输入，而客户端缓慢的处理速度（当然也可能是一个拥塞的网络环境）使得服务器花费了不少等待时间。

我们可以试想一下，如果服务器要处理大量的请求连接，每个请求如果都像这样拖慢了服务器的处理速度，那么服务端能够处理的并发数量就会大幅度减少。反之，如果服务器每次都能很快地处理一次请求，那么相对的，它的并发能力就能上升。

在这个案例中，服务器处理请求之所以慢，并不是因为在服务端有多少繁重的任务，而仅仅是因为服务线程在等待IO而已。让高速运转的CPU去等待极其低效的网络IO是非常不合算的行为。那么，我们是不是可以想一个方法，将网络IO的等待时间从线程中分离出来呢？

5.10.2　使用NIO进行网络编程

使用Java的NIO就可以将上面的网络IO等待时间从业务处理线程中抽取出来。那么NIO是什么，它又是如何工作的呢？

要了解NIO，我们首先需要知道在NIO中的一个关键组件Channel（通道）。Channel有点类似于流，一个Channel可以和文件或者网络Socket对应。如果Channel对应着一个Soceket，那么往这个Channel中写数据，就等同于向Socket中写入数据。

和Channel一起使用的另外一个重要组件就是Buffer。大家可以简单地把Buffer理解成一个内存区域或者byte数组。数据需要包装成Buffer的形式才能和Channel交互（写入或者读取）。

另外一个与Channel密切相关的是Selector（选择器）。在Channel的众多实现中，有一个SelectableChannel实现，表示可被选择的通道。任何一个SelectableChannel都可以将自己注册到一个Selector中。这样，这个Channel就能被Selector所管理。而一个Selector可以管理多个SelectableChannel。当SelectableChannel的数据准备好时，Selector就会接到通知，得到那些已经准备好的数据。而SocketChannel就是SelectableChannel的一种。因此，它们构成了如图5.20所示的结构。

图5.20　Selector和Channel

大家可以看到，一个Selector可以由一个线程进行管理，而一个SocketChannel则可以表示一个客户端连接，因此这就构成由一个或者极少数线程，来处理大量客户端连接的结构。当与客户端连接的数据没有准备好时，Selector会处于等待状态（不过幸好，用于管理Selector的线程数是极少量的），而一旦有任何一个SocketChannel准备好了数据，Selector就能立即得到通知，获取数据进行处理。

下面就让我们用NIO来重新构造这个多线程的Echo服务器吧！

首先，我们需要定义一个Selector和线程池：

private Selector selector;
private ExecutorService  tp=Executors.newCachedThreadPool();

其中，selector用于处理所有的网络连接。线程池tp用于对每一个客户端进行相应的处理，每一个请求都会委托给线程池中的线程进行实际的处理。

为了能够统计服务器线程在一个客户端上花费了多少时间，这里还需要定义一个与时间统计有关的类：

public static Map＜Socket,Long＞ time_stat=new HashMap＜Socket,Long＞(10240);

它用于统计在某一个Socket上花费的时间，time_stat的key为Socket，value为时间戳（可以记录处理开始时间）。

下面就可以来看一下NIO服务器的核心代码，下面的startServer()方法用于启动NIO Server：

01 private void startServer() throws Exception {
02   selector = SelectorProvider.provider().openSelector();
03   ServerSocketChannel ssc = ServerSocketChannel.open();
04   ssc.configureBlocking(false);
05
06   InetSocketAddress isa = new InetSocketAddress(InetAddress.getLocalHost(), 8000);
07   InetSocketAddress isa = new InetSocketAddress(8000);
08   ssc.socket().bind(isa);
09
10   SelectionKey acceptKey = ssc.register(selector, SelectionKey.OP_ACCEPT);
11
12   for (;;) {
13     selector.select();
14     Set readyKeys = selector.selectedKeys();
15     Iterator i = readyKeys.iterator();
16     long e=0;
17     while (i.hasNext()) {
18       SelectionKey sk = (SelectionKey) i.next();
19       i.remove();
20
21       if (sk.isAcceptable()) {
22         doAccept(sk);
23       }
24       else if (sk.isValid() && sk.isReadable()) {
25         if(!time_stat.containsKey(((SocketChannel)sk.channel()).socket()))
26           time_stat.put(((SocketChannel)sk.channel()).socket(),
27             System.currentTimeMillis());
28         doRead(sk);
29       }
30       else if (sk.isValid() && sk.isWritable()) {
31         doWrite(sk);
32         e=System.currentTimeMillis();
33         long b=time_stat.remove(((SocketChannel)sk.channel()).socket());
34         System.out.println("spend:"+(e-b)+"ms");
35       }
36     }
37   }
38 }

上述代码第2行，通过工厂方法获得一个Selector对象的实例。第3行，获得表示服务端的SocketChannel实例。第4行，将这个SocketChannel设置为非阻塞模式。实际上，Channel也可以像传统的Socket那样按照阻塞的方式工作。但在这里，更倾向于让其工作在非阻塞模式，在这种模式下，我们才可以向Channel注册感兴趣的事件，并且在数据准备好时，得到必要的通知。接着，在第6～8行进行端口绑定，将这个Channel绑定在8000端口。

在第10行，将这个ServerSocketChannel绑定到Selector上，并注册它感兴趣的时间为Accept。这样，Selector就能为这个Channel服务了。当Selector发现ServerSocketChannel有新的客户端连接时，就会通知ServerSocketChannel进行处理。方法register()的返回值是一个SelectionKey，SelectionKey表示一对Selector和Channel的关系。当Channel注册到Selector上时，就相当于确立了两者的服务关系，那么SelectionKey就是这个契约。当Selector或者Channel被关闭时，它们对应的SelectionKey就会失效。

第12～37行是一个无穷循环，它的主要任务就是等待-分发网络消息。

第13行的select()方法是一个阻塞方法。如果当前没有任何数据准备好，它就会等待。一旦有数据可读，它就会返回。它的返回值是已经准备就绪的SelectionKey的数量。这里简单地将其忽略。

第14行获取那些准备好的SelectionKey。因为Selector同时为多个Channel服务，因此已经准备就绪的Channel就有可能是多个。所以，这里得到的自然是一个集合。得到这个就绪集合后，剩下的就是遍历这个集合，挨个处理所有的Channel数据。

第15行得到这个集合的迭代器。第17行使用迭代器遍历整个集合。第18行根据迭代器获得一个集合内的SelectionKey实例。

第19行将这个元素移除！注意，这个非常重要，否则就会重复处理相同的SelectionKey。当你处理完一个SelectionKey后，务必将其从集合内删除。

第21行判断当前SelectionKey所代表的Channel是否在Acceptable状态，如果是，就进行客户端的接收（执行doAccept()方法）。

第24行判断Channel是否已经可以读了，如果是就进行读取（doRead()方法）。这里为了统计系统处理每一个连接的时间，在第25～27行记录了在读取数据之前的一个时间戳。

第30行判断通道是否准备好进行写。如果是就进行写入（doWrite()方法），同时在写入完成后，根据读取前的时间戳，输出处理这个Socket连接的耗时。

在了解服务端的整体框架后，下面让我们从细节着手，学习一下几个主要方法的内部实现。首先是doAccept()方法，它与客户端建立连接：

01 private void doAccept(SelectionKey sk) {
02   ServerSocketChannel server = (ServerSocketChannel) sk.channel();
03   SocketChannel clientChannel;
04   try {
05     clientChannel = server.accept();
06     clientChannel.configureBlocking(false);
07
08     // Register this channel for reading.
09     SelectionKey clientKey = clientChannel.register(selector, SelectionKey.OP_READ);
10     // Allocate an EchoClient instance and attach it to this selection key.
11     EchoClient echoClient = new EchoClient();
12     clientKey.attach(echoClient);
13
14     InetAddress clientAddress = clientChannel.socket().getInetAddress();
15     System.out.println("Accepted connection from " + clientAddress.getHostAddress() + ".");
16   } catch (Exception e) {
17     System.out.println("Failed to accept new client.");
18     e.printStackTrace();
19   }
20 }

和Socket编程很类似，当有一个新的客户端连接接入时，就会有一个新的Channel产生代表这个连接。上述代码第5行，生成的clientChannel就表示和客户端通信的通道。第6行，将这个Channel配置为非阻塞模式，也就是要求系统在准备好IO后，再通知我们的线程来读取或者写入。

第9行很关键，它将新生成的Channel注册到selector选择器上，并告诉Selector，我现在对读（OP_READ）操作感兴趣。这样，当Selector发现这个Channel已经准备好读时，就能给线程一个通知。

第11行新建一个对象实例，一个EchoClient实例代表一个客户端。在第12行，我们将这个客户端实例作为附件，附加到表示这个连接的SelectionKey上。这样在整个连接的处理过程中，我们都可以共享这个EchoClient实例。

EchoClient的定义很简单，它封装了一个队列，保存在需要回复给这个客户端的所有信息，这样，再进行回复时，只要从outq对象中弹出元素即可。

class EchoClient {
  private LinkedList＜ByteBuffer＞ outq;
  EchoClient() {
    outq = new LinkedList＜ByteBuffer＞();
  }
  public LinkedList＜ByteBuffer＞ getOutputQueue() {
    return outq;
  }
  public void enqueue(ByteBuffer bb) {
    outq.addFirst(bb);
  }
}

下面来看一下另外一个重要的方法doRead()。当Channel可以读取时，doRead()方法就会被调用。

01 private void doRead(SelectionKey sk) {
02   SocketChannel channel = (SocketChannel) sk.channel();
03   ByteBuffer bb = ByteBuffer.allocate(8192);
04   int len;
05
06   try {
07     len = channel.read(bb);
08     if (len ＜ 0) {
09       disconnect(sk);
10       return;
11     }
12   } catch (Exception e) {
13     System.out.println("Failed to read from client.");
14     e.printStackTrace();
15     disconnect(sk);
16     return;
17   }
18
19   bb.flip();
20   tp.execute(new HandleMsg(sk,bb));
21 }

方法doRead()接收一个SelectionKey参数，通过这个SelectionKey可以得到当前的客户端Channel（第2行）。在这里，我们准备8K的缓冲区读取数据，所有读取的数据存放在变量bb中（第7行）。读取完成后，重置缓冲区，为数据处理做准备（第19行）。

在这个示例中，我们对数据的处理很简单。但是为了模拟复杂的场景，还是使用了线程池进行数据处理（第20行）。这样，如果数据处理很复杂，就能在单独的线程中进行，而不用阻塞任务派发线程。

HandleMsg的实现也很简单：

01 class HandleMsg implements Runnable{
02   SelectionKey sk;
03   ByteBuffer bb;
04   public HandleMsg(SelectionKey sk,ByteBuffer bb){
05     this.sk=sk;
06     this.bb=bb;
07   }
08   @Override
09   public void run() {
10     EchoClient echoClient = (EchoClient) sk.attachment();
11     echoClient.enqueue(bb);
12     sk.interestOps(SelectionKey.OP_READ | SelectionKey.OP_WRITE);
13     //强迫selector立即返回
14     selector.wakeup();
15   }
16 }

上述代码，简单地将接收到的数据压入EchoClient的队列（第11行）。如果需要处理业务逻辑，就可以在这里进行处理。

在数据处理完成后，就可以准备将结果回写到客户端，因此，重新注册感兴趣的消息事件，将写操作（OP_WRITE）也作为感兴趣的事件进行提交（第12行）。这样在通道准备好写入时，就能通知线程。

写入操作使用doWrite()函数实现：

01 private void doWrite(SelectionKey sk) {
02   SocketChannel channel = (SocketChannel) sk.channel();
03   EchoClient echoClient = (EchoClient) sk.attachment();
04   LinkedList＜ByteBuffer＞ outq = echoClient.getOutputQueue();
05
06   ByteBuffer bb = outq.getLast();
07   try {
08     int len = channel.write(bb);
09     if (len == -1) {
10       disconnect(sk);
11       return;
12     }
13
14     if (bb.remaining() == 0) {
15       // The buffer was completely written, remove it.
16       outq.removeLast();
17     }
18   } catch (Exception e) {
19     System.out.println("Failed to write to client.");
20     e.printStackTrace();
21     disconnect(sk);
22   }
23
24   if (outq.size() == 0) {
25     sk.interestOps(SelectionKey.OP_READ);
26   }
27 }

函数doWrite()也接收一个SelectionKey，当然针对一个客户端来说，这个SelectionKey实例和doRead()拿到的SelectionKey是同一个。因此，通过SelectionKey我们就可以在这两个操作中共享EchoClient实例。上述代码第3～4行，我们取得EchoClient实例以及它的发送内容列表。第6行，获得列表顶部元素，准备写回客户端。第8行进行写回操作。如果全部发送完成，则移除这个缓存对象（第16行）。

在doWrite()中最重要的，也是最容易被忽略的是在全部数据发送完成后（也就是outq的长度为0），需要将写事件（OP_WRITE）从感兴趣的操作中移除（第25行）。如果不这么做，每次Channel准备好写时，都会来执行doWrite()方法。而实际上，你又无数据可写，这显然是不合理的。因此，这个操作很重要。

上面我们已经介绍了主要的核心代码，现在使用这个NIO服务器来处理上一节中客户端的访问。同样的，客户端也是要花费将近6秒钟，才能完成一次消息的发送，那使用NIO技术后，服务端线程需要花费多少时间来处理这些请求呢？答案如下：

spend:2ms
spend:2ms
spend:2ms
spend:2ms
spend:3ms
spend:3ms
spend:0ms
spend:0ms
spend:2ms
spend:3ms

可以看到，在使用NIO技术后，即使客户端迟钝或者出现了网络延迟等现象，并不会给服务器带来太大的问题。

5.10.3　使用NIO来实现客户端

在前面的案例中，我们使用Socket编程来构建我们的客户端，使用NIO来实现服务端。实际上，使用NIO也可以用来创建客户端。这里，我们再演示一下使用NIO创建客户端的例子。

和构造服务器类似，核心的元素也是Selector、Channel和SelectionKey。

首先，我们需要初始化Selector和Channel：

01 private Selector selector;
02 public void init(String ip, int port) throws IOException {
03   SocketChannel channel = SocketChannel.open();
04   channel.configureBlocking(false);
05   this.selector = SelectorProvider.provider().openSelector();
06   channel.connect(new InetSocketAddress(ip, port));
07   channel.register(selector, SelectionKey.OP_CONNECT);
08 }

上述代码第3行，创建一个SocketChannel实例，并设置为非阻塞模式。第5行创建了一个Selector。第6行，将SocketChannel绑定到Socket上。但由于当前Channel是非阻塞的，因此，connect()方法返回时，连接并不一定建立成功，在后续使用这个连接时，还需要使用finishConnect()再次确认。第7行，将这个Channel和Selector进行绑定，并注册了感兴趣的事件作为连接（OP_CONNECT）。

初始化完成后，就是程序的主要执行逻辑：

01 public void working() throws IOException {
02   while (true) {
03     if (!selector.isOpen())
04       break;
05     selector.select();
06     Iterator＜SelectionKey＞ ite = this.selector.selectedKeys().iterator();
07     while (ite.hasNext()) {
08       SelectionKey key = ite.next();
09       ite.remove();
10       // 连接事件发生
11       if (key.isConnectable()) {
12         connect(key);
13       } else if (key.isReadable()) {
14         read(key);
15       }
16     }
17   }
18 }

在上述代码中，第5行通过Selector得到已经准备好的事件。如果当前没有任何事件准备就绪，这里就会阻塞。这里的整个处理机制和服务端非常类似，主要处理两个事件，首先是表示连接就绪的Connct事件（由connect()函数处理）以及表示通道可读的Read事件（由read()函数处理）。

函数connect()的实现如下：

01 public void connect(SelectionKey key) throws IOException {
02   SocketChannel channel = (SocketChannel) key.channel();
03   // 如果正在连接，则完成连接
04   if (channel.isConnectionPending()) {
05     channel.finishConnect();
06   }
07   channel.configureBlocking(false);
08   channel.write(ByteBuffer.wrap(new String("hello server!\r\n")
09       .getBytes()));
10   channel.register(this.selector, SelectionKey.OP_READ);
11 }

上述connect()函数接收SelectionKey作为其参数。在第4～6行，它首先判断是否连接已经建立，如果没有，则调用finishConnect()完成连接。建立连接后，向Channel写入数据，并同时注册读事件为感兴趣的事件（第10行）。

当Channel可读时，会执行read()方法，进行数据读取：

01 public void read(SelectionKey key) throws IOException {
02   SocketChannel channel = (SocketChannel) key.channel();
03   // 创建读取的缓冲区
04   ByteBuffer buffer = ByteBuffer.allocate(100);
05   channel.read(buffer);
06   byte[] data = buffer.array();
07   String msg = new String(data).trim();
08   System.out.println("客户端收到信息：" + msg);
09   channel.close();
10   key.selector().close();
11 }

上述read()函数首先创建了100字节的缓冲区（第4行），接着从Channel中读取数据，并将其打印在控制台上。最后，关闭Channel和Selector。