Java NIO 详解

 

 

Java NIO 详解

2010-08-30 18:17 1376人阅读 评论(4) 收藏 举报

 

前记：

       有一个周末的晚上，我突然想阅读以前写过的一些引以为自豪的程序；于是乎，我就把 9 个月前做过的一个项目的源码翻出来读了，这是一个 Socket 项目，服务器端是用 Java 实现的，客户端是用 Flash 实现的；这个项目的目的就是需要保证服务器端和客户端的连接不掉链，并且如果客户端有心跳超时或者连接断开的情形 , 我们会立即通知另一个应用程序（ probe ）。

    这个程序本身的业务非常简单，也就是：与客户端建立连接，客户端发送事件数据，转发事件数据，检测连接状态，通知 probe ，监控等。然而因为这是一个网络编程并且又是 socket ，而且还要支持大并发操作，当然从技术上来讲还是有一点难度的；直观的印象肯定是服务器端肯定需要用到很多线程池，流操作，字符操作，移位操作，锁操作等比较高级的技术；确实，我们在开发这个项目的时候，为了权衡高并发和数据一致性方面做了很多的工作，利用各种各样的锁。

    当时，我也是第一次写这样的程序，这个项目做完之后我还是挺有成就感的。因为从中我学会很多很复杂所谓很高级的技术。 9 个月之后的这个晚上，当我再次拜读这个项目的代码的时候，我有一种莫名的忧伤，我似乎再也不可能以此段代码来引以自豪了。代码本身没有什么漏洞和 bug ，但从架构上和效率上来讲实在是有点低下了，程序中应用了大量的锁，大量的轮询，大量的阻塞等待，大量的线程休眠操作，这些操作都会浪费资源并且导致效率的低下。

.

 

      上图 就是这个项目的总体结构图，从图中可以看出该程序分为这几大块：连接侦听线程、连接对象队列、发送线程池、接收线程池、分发线程、事件处理对象、监控处理对象。下面我将描述下整个连接处理过程：

1、 连接侦听线程循环接收一个连接请求，如果有连接请求过来，则返回一个连接 Socket 对象，否则该线程就阻塞等待，直到有一个连接请求过来。

2、 封装该返回的 Socket 对象（主要是封装获取完整包数据，发送方法，关闭方法等）成 Connection 对象，并把封装好的 Connection 对象放入连接对象队列。

3、 分发线程不停的轮询连接对象队列，如果发现有可接收数据的连接对象，则扔给接收线程池去处理；如果发现有可发送数据的连接对象，则扔给发送线程池去处理。如果轮询一圈发现既没有可发送数据的连接对象也没有可接收数据的连接对象，则该线程会休眠一段时间，休眠过后又接着循环。

4、 发送线程池内有一个连接对象队列，从队列中取出一个连接对象并发送数据，且记录连接状态信息。

5、 接收线程池内也有一个连接对象队列，从队列中取出一个连接对象并接收一个数据包，且记录连接状态信息。如果接收的数据包是心跳检测包则更新连接状态，如果是数据包则通过事件处理对象发送给 probe 系统。

    从上面的过程来看，我们可能看不出设计上面的漏洞，但有几个地方确实非常影响效率，在这里我想先提出来：

1、 连接侦听线程一直在侦听，有连接请求过来则会返回，没有则会阻塞；这样这个线程就会一直挂着；如果时时刻刻都有很多的连接过来，这个线程还会充分发挥它的作用，但其实大部分时候，连接请求并没有这么频繁，所以这个线程大部分时间是阻塞的；这样为了这样一个功能单独利用一个线程就有点浪费了。

2、 分发线程不停的轮询过程是导致整个系统效率低下最严重的一块，分发线程不停的轮询连接对象队列，其实分发线程并不知道哪个线程需要发送数据，哪些线程需要接收数据，而他只是盲目地从队列的头遍历到队列的尾部，如果发现没有可操作的连接对象则休眠一段时间；其实在大部分情况下，连接对象并不是时时刻刻都有数据发送和接收，所以这个分发线程大部分时间空循环，白忙了；并且这个休眠时间也不好控制，如果时间长了，则程序的即时性不够，如果太短了，程序似乎就是在空跑了。

3、 在连接对象上发送和接收数据包的时候，这些方法都是阻塞操作的；所以当有大量的数据可接收和发送的时候，这种阻塞的操作时非常浪费资源的。

    以上所提出的问题，如果是在并发规模比较小的情况下，是没有什么问题；但确实有很大的改进空间。上面的问题归结起来主要是两个：

1、 当有连接请求过来或者有 Socket 连接有数据可读可写的时候，我们不会立即知道，我们必须要一个一个的轮询，我们能否有一种机制，即是，当有连接请求过来或者连接有数据可读或者可写的时候，直接通知我们来处理，而不需要我们主动轮询。

2、 当读数据或者写数据的时候，所有的方法都阻塞了，能不能有一种办法当我们写数据或者接收数据的时候不用阻塞，而是直接返回，这样就明显提高了线程的使用率了。

    值得我们庆幸的是，在 Java 的 JDK1.4 之后的版本，提供了 NIO 包，这里提出了事件驱动的 I/O 编程模式和非阻塞信道的概念， NIO 里面的 Selector 对象解决了上面提出分发和轮询的问题， Channel 接口解决了阻塞读写的问题。我相信这些组件能够帮我们解决上面所提出的所有问题。所以下面有很大一部分篇幅来介绍NIO 的使用和一些底层的机制。

 

NIO 详解：

    NIO 主要包括两个部分： java.nio.channels 包介绍 Selector 和 Channel 抽象， java.nio 包介绍Buffer 抽象。这都是一些高级的特性，有许多微妙的使用细节。但是，我在下面不会对 Buffer 抽象上介绍太多，如果想要更深入的了解 Buffer 对象请阅读相关的书籍和资料。

1、 为什么需要 NIO

基本的 Java 套接字对于小规模系统可以很好地运行，但当涉及同时处理几千甚至上万个客户端的服务器时，可能会产生一些问题。如果一个客户端一个线程的方式去处理，则由于创建、维护和切换线程需要的系统开销导致系统扩展性方面受到了很大限制；当然你也可以使用线程池，也可以节省一些开销，也同时可以使用并行硬件的优势，比如 F5 ，网络连接均衡服务器等等。但对于连接生存期比较长的协议来说，线程池的大小仍然限制了系统可以同时处理的客户端数量。考虑一个在客户端之间传递消息的即时消息服务器 IM 。客户端必须不停地连接服务器以接收即时消息，因此线程池的大小限制了系统可以同时服务的客户端总数。如果增加线程池的大小，将带来更多的线程处理开销，而不能提升系统的性能，因为在大部分的时间里客户端是处于空闲状态的。

如果这是所有问题，可能 NIO 还不是必须的。不幸的是，在使用线程的扩展性方面还涉及一些更加难把握的挑战。其中一个挑战就是程序员几乎不能对什么时候哪个线程将获得服务进行控制。你可以设置一个线程实例的优先级，但是这个优先级只是一种“建议”，下一个选择执行的线程完全取决于具体实现。因此，如果程序员想要保证某些连接优先获得服务，或想要制定一定的服务顺序，线程可能就很难做到。

然而，有关线程的最重要的问题可能要保证数据的一致性，但很多客户端之间共享一些状态信息时，这就需要使用锁机制或者其他互斥机制对依次访问状态进行严格的同步。否则，由于不同线程上的程序段交错执行，他们之间会改掉其他线程说做的修改。

由于需要对共享状态进行同步访问，要同时考虑到多线程服务器的正确性和高效性就变得非常困难。使用锁机制将增加更多的的系统调度和上下文切换开销，而程序员对这些开销又无法控制。由于其复杂性，一些程序员宁愿继续使用单线程方法。这类服务器只用一个线程来处理所有客户端，但不是顺序处理，而是一次全部处理。这种服务器不能为任何客户端提供 I/O 操作的阻塞等待，而必须排他地使用非阻塞 I/O 。

在我们写 Socket 服务器端的时候，肯定会用到 ServerSocket 类的 accept 方法，当在 ServerSocket实例上调用 accept 方法时，如果有一个新的连接来了，则 accept 方法会立即返回一个 socket 实例，否则该方法将一直阻塞直到有新的连接到来或计时器超时。假设我们用一个线程专门来处理连接的请求，也就是accept 方法；不幸的是，我们会发现这种方法要求我们不断地轮询所有的 I/O 源，而这种“忙等”方法又会引入很多系统开销，因为程序要反复循环地连接 I/O 源，却又发现什么都不用做。以下代码就是一个典型的处理客户端请求方式，循环一直在跑，除非有人把循环标志给修改了， server.accept() 方法一直在阻塞直到有一个新的的连接过来，如果有新的连接过来这返回一个 socket 实例，并扔给连接管理器去处理，如果一直都没有连接过来则一直阻塞在那里死等。

while (! bCanExit ) {

try {

           // 该方法一直会阻塞，直到有新的连接过来

          Socket socket = server .accept();

         Connection connection = new Connection(socket);

         connection.setClientId(Util.random32UUID ());

         connectionManager .add(connection);

         if ( logger .isInfoEnabled()){

              logger .info( " 有一个新的连接 !" );

         }

} catch (IOException e) {}

}

try {

server .close();

} catch (IOException e) {

logger. error( "close serverSocket error:" , e);

}

我们需要一种方法来一次轮询一组客户端，以查找哪个客户端需要服务。这正是 NIO 中将要介绍的Selector 和 Channel 抽象的关键的。一个 Channel 实例代表了一个“可轮询的” I/O 目标，如套接字（或一个文件、设备等）。 Channel 能够注册一个 Selector 类的实例。 Selector 的 select() 方法允许你询问“在一组信道中，哪一个当前需要服务（即，被接收，读或写）” , 这两个类都包含在 java.nio.channels包中。

NIO 中的另一个主要特性是 Buffer 类。就像 selector 和 channel 为一次处理多个客户端的系统开销提供了更高级的控制和可预测性。 Buffer 则提供了比 Stream 抽象更高效和可预测的 I/O 。 Stream 抽象好的方面是隐藏了底层缓冲区的有限性，提供了一个能够容纳任意长度数据的容器的假象。坏的方面是要实现这样一个假象，要么产生大量的内存开销，要么会引入大量的上下文切换，甚至可能两者都有。在使用线程时，这些开销都隐藏在具体实现中，因此也失去了对其可控性和可预测性。这种方法使编写程序变得容易，但要调整它们的性能则变得更困难。不幸的是，如果要使用 Java 的 Socket 抽象，流就是唯一的选择。

这就是为什么要把 channel 设计为使用 Buffer 实例来传递数据的原因。 Buffer 抽象代表了一个有限容量的数据容器（其本质是一个数组），由指针指示了在哪存放数据和从哪里读取数据。使用 Buffer 有两个主要好处。第一，与读写缓冲区数据相关联的系统开销暴露给了程序员。例如，如果想要向缓冲区写入数据，但又没有足够的空间时，就必须采取一些措施来获取空间（即移出一些数据，或移开已经在那个位置的数据来获得空间，或者创建一个新的实例）。这意味着需要额外的工作，但是你可以控制它什么时候发生，如何发生，以及是否发生。一个聪明的程序员如果清楚地了解了应用程序的需求，就那能通过权衡这些选择来降低系统开销。第二，一些 Java 对象的特殊 buffer 映射操作能够直接操作底层平台的资源（如操作系统的缓冲区）。这些操作节省了在不同地址空间中复制数据的开销（这在现代计算机体系结构中是开销很大的操作）。

2、 与 Buffer 一起使用 Channel

前面已经讲过， Channel 实例代表了一个与设备（或文件）通信的通道，通过它可以进行输入输出操作。实际上 Channel 的基本思想与我们见过的普通套接字非常相似。对于 TCP 协议，可以使用ServerSocketChannel 和 SocketChannel 。信道（ Channel ）和套接字（ Socket ）之间的不同点之一，可能是信道通常要调用静态工厂方法来获取实例，如：

SocketChannel clntChan = SocketChannel.open ();

ServerSocketChannel serverChan = ServerSocketChannel.open();

Channel 使用的不是流，而是用缓冲区来发送或读取数据。 Buffer 类或其任何子类的实例都可以看作是一个定长的 Java 基本数据类型元素序列。与流不同，缓冲区有固定的、有限的容量，并由内部（但可以被访问）状态记录了有多少数据放入或取出，就像是有限容量的队列一样。 Buffer 是一个抽象类，只能通过创建它的子类来获取 Buffer 实例，而每个子类都设计为用来容纳一种 Java 基本数据类型（ boolean 除外）。因此，这些实例分别为 FloatBuffer 、 IntBuffer 或 ByteBuffer 等。在 channel 中使用 Buffer 实例通常不是使用构造函数创建的，而是通过调用 allocate() 方法创建指定容量的 Buffer 实例，如：

ByteBuffer buffer = ByteBuffer.allocate(bodyLength);

或者通过包装一个已有的数组来创建：

ByteBuffer buffer = ByteBuffer.wrap(byteArray);

NIO 的强大功能部分来自于 channel 的非阻塞特性。上面介绍的内容可以知道，套接字的某些操作可能会无限期地阻塞。例如，对 accept() 方法的调用可能会因为等待一个客户端连接而阻塞；对 read() 方法的调用可能会因为没有数据可读而阻塞，直到连接的另一端传来新的数据。总的来说，创建 / 接收连接或读写数据等 I/O 调用，都可能无限期地阻塞等待，直到底层的网络实现发生了什么。慢速的，有损耗的网络，或仅仅是简单的网络故障都可能导致任意时间的延迟。然而不幸的是，在调用一个方法之前无法知道其是否阻塞。 NIO 的 channel抽象的一个重要特征就是可以通过配置它的阻塞行为，以实现非阻塞式的信道。

clntChan.configureBlocking(false);

    在非阻塞式信道上调用一个方法总是会立即返回。这种调用的返回值指示了所请求的操作完成的程度。例如，在一个非阻塞式 ServerSocketChannel 上调用 accept() 方法，如果有连接请求来了，则返回客户端SocketChannel ，否则返回 null 。

    下面来创建一个非阻塞式 TCP 回显客户端（就是客户端发送什么数据到服务器端，服务器端就返回什么数据回来）。可能阻塞的 I/O 操作包括建立连接，读和写。通过使用非阻塞式信道，这些操作都将立即返回，我们必须反复调用这些操作，直到所有 I/O 操作都成功完成。

客户端代码：

package org.zapldy.tcpip.nio;

 

import java.net.InetSocketAddress;

import java.net.SocketException;

import java.nio.ByteBuffer;

import java.nio.channels.SocketChannel;

 

public class TCPEchoClientNoblocking {

 

    public static void main(String[] args) throws Exception {

       String server = "127.0.0.1" ;

       byte [] data = "ABCDEFGHIJKLMNOPQRSTUVWXYZ" .getBytes();

       int servPort = 8888;

 

       SocketChannel clntChan = SocketChannel.open ();

       clntChan.configureBlocking( false );

       // 我们通过持续调用 finishConnect() 方法来“轮询”连接状态，该方法在连接成功建立之前

        // 一直返回 false 。打印操作显示了在等待连接建立的过程中，程序还可以执行其他任务。不过

        //, 这种忙等的方法非常浪费系统资源，这里这样做只是为了演示该方法的使用。

       if (!clntChan.connect( new InetSocketAddress(server, servPort))) {

           while (!clntChan.finishConnect()) {

              System. out .print( "=" ); // 这里可以做其他事情

           }

       }

 

       ByteBuffer writeBuf = ByteBuffer.wrap (data);

       ByteBuffer readBuf = ByteBuffer.allocate (data. length );

 

       int totalBytesRcvd = 0;

       int bytesRcvd;

       while (totalBytesRcvd < data. length ) {

           if (writeBuf.hasRemaining()) {

              clntChan.write(writeBuf);

           }

           if ((bytesRcvd = clntChan.read(readBuf)) == -1) {

              throw new SocketException( "Connection closed prematurely" );

           }

           totalBytesRcvd += bytesRcvd;

           System. out .print( "=" );

       }

       System. out .println( "Recieved: "

              + new String(readBuf.array(), 0, totalBytesRcvd));

       clntChan.close();

    }

}

    服务器端代码（服务器还是用传统的方式实现的，后面将再用 NIO 重写，这里只是为了读者能快速运行代码）：

package org.zapldy.tcpip;

 

import java.io.IOException;

import java.io.InputStream;

import java.io.OutputStream;

import java.net.ServerSocket;

import java.net.Socket;

import java.net.SocketAddress;

 

public class TCPEchoServer {

    private static final int BUFSIZE = 32;

    public static void main(String[] args) throws IOException {

       int servPort = 8888;

       ServerSocket servSocket = new ServerSocket(servPort);

 

       int recvMsgSize = 0;

       byte [] recvBuf = new byte [ BUFSIZE ];

 

       while ( true ) {

           Socket clntSocket = servSocket.accept(); // 该方法会阻塞

           SocketAddress clientAddress =

                                clntSocket.getRemoteSocketAddress();

           System. out .println( "Handling client at " + clientAddress);

 

           InputStream in = clntSocket.getInputStream();

           OutputStream out = clntSocket.getOutputStream();

 

           while ((recvMsgSize = in.read(recvBuf)) != -1) {

              out.write(recvBuf, 0, recvMsgSize);

           }

           clntSocket.close();

       }

    }

}

 

3、 Selector

上面已经提到过， Selector 类可以用于避免使用阻塞式客户端中很浪费资源的“忙等”方法。例如，考虑一个 IM 服务器。像 QQ 或者旺旺这样的，可能有几万甚至几千万个客户端同时连接到了服务器，但在任何时刻都只是非常少量的消息

需要读取和分发。这就需要一种方法阻塞等待，直到至少有一个信道可以进行 I/O 操作，并指出是哪个信道。 NIO 的选择器就实现了这样的功能。一个 Selector 实例可以同时检查一组信道的 I/O 状态。用专业术语来说，选择器就是一个多路开关选择器，因为一个选择器能够管理多个信道上的 I/O 操作。然而如果用传统的方式来处理这么多客户端，使用的方法是循环地一个一个地去检查所有的客户端是否有 I/O 操作，如果当前客户端有 I/O 操作，则可能把当前客户端扔给一个线程池去处理，如果没有 I/O 操作则进行下一个轮询，当所有的客户端都轮询过了又接着从头开始轮询；这种方法是非常笨而且也非常浪费资源，因为大部分客户端是没有 I/O 操作，我们也要去检查；而 Selector 就不一样了，它在内部可以同时管理多个 I/O ，当一个信道有 I/O 操作的时候，他会通知 Selector ， Selector 就是记住这个信道有 I/O 操作，并且知道是何种 I/O 操作，是读呢？是写呢？还是接受新的连接；所以如果使用 Selector ，它返回的结果只有两种结果，一种是 0 ，即在你调用的时刻没有任何客户端需要 I/O 操作，另一种结果是一组需要 I/O 操作的客户端，这是你就根本不需要再检查了，因为它返回给你的肯定是你想要的。这样一种通知的方式比那种主动轮询的方式要高效得多！

要使用选择器（ Selector ），需要创建一个 Selector 实例（使用静态工厂方法 open() ）并将其注册（register ）到想要监控的信道上（注意，这要通过 channel 的方法实现，而不是使用 selector 的方法）。最后，调用选择器的 select() 方法。该方法会阻塞等待，直到有一个或更多的信道准备好了 I/O 操作或等待超时。 select() 方法将返回可进行 I/O 操作的信道数量。现在，在一个单独的线程中，通过调用 select() 方法就能检查多个信道是否准备好进行 I/O 操作。如果经过一段时间后仍然没有信道准备好， select() 方法就会返回 0 ，并允许程序继续执行其他任务。

下面来看一个例子。假设我们想要使用信道和选择器来实现一个像上面一样的回显服务器，并不使用多线程和忙等。为了使不同协议都能方便地使用这个基本的服务模式，我们把信道中与具体协议相关的处理各种 I/O 操作（接收，读，写）分离出来。 Protocol 定义了通用 EchoSelectorServer 类与特定协议之间的接口，包括三个方法，每个方法代表了一种 I/O 形式。当有信道准备好 I/O 操作时，服务器只需要调用相应的方法即可。

package org.zapldy.tcpip.nio;

 

import java.io.IOException;

import java.nio.channels.SelectionKey;

 

public interface Protocol {

    public void handleAccept(SelectionKey key) throws IOException;

    public void handleRead(SelectionKey key) throws IOException;

    public void handleWrite(SelectionKey key) throws IOException;

}

下面是具体的实现（注意看注释）：

package org.zapldy.tcpip.nio;

 

import java.io.IOException;

import java.nio.ByteBuffer;

import java.nio.channels.SelectionKey;

import java.nio.channels.ServerSocketChannel;

import java.nio.channels.SocketChannel;

 

public class EchoProtocol implements Protocol {

 

    private int bufsize ; // 为每个客户端信道创建的缓冲区大小

 

    public EchoProtocol( int bufsize) {

       this . bufsize = bufsize;

    }

 

 

    public void handleAccept (SelectionKey key) throws IOException {

       //channel() 方法返回注册时用来创建的 Channel ，该 Channel 是一个 ServerSocketChannel ，

        // 因为这是我们注册的唯一一种支持 accept 操作的信道，

//accept() 方法为传入的连接返回一个 SocketChannel 实例。

       SocketChannel channel = ((ServerSocketChannel)           key.channel()).accept();

       // 这里无法注册阻塞式信道，必须是非阻塞式的

       channel.configureBlocking( false );

       // 可以通过 SelectionKey 类的 selector() 方法来获取相应的 Selector 。

        // 我们根据指定大小创建了一个新的 ByteBuffer 实例，

        // 并将其作为参数传递给 register() 方法。它将作为附件，与 regiter() 方法所返回的

//SelectionKey 实例相关联。

       channel.register(key.selector(), SelectionKey. OP_READ , ByteBuffer

              .allocateDirect ( bufsize ));

    }

 

   

    public void handleRead(SelectionKey key) throws IOException {

       // 根据其支持数据读取操作可知，这是一个 SocketChannel 。

       SocketChannel channel = (SocketChannel) key.channel();

       // 建立连接后，有一个 ByteBuffer 附件加到该 SelectionKey 实例上，这个附件里面的内容将

        // 会在发送的时候用到，附件始终是附着这个长连接上

       ByteBuffer buf = (ByteBuffer) key.attachment();

       long bytesRead = channel.read(buf);

       // 如果 read() 方法返回 -1 ，则表示底层连接已经关闭，此时需要关闭信道。

        // 关闭信道时，将从选择器的各种集合中移除与该信道关联的键。

       if (bytesRead == -1) {

           channel.close();

       } else if (bytesRead > 0) {

           // 这里依然保留了信道的可读操作，虽然缓冲区中可能已经没有剩余空间了，

// 因为下次还是要接受新的数据

           key.interestOps(SelectionKey. OP_READ | SelectionKey. OP_WRITE );

       }

    }

 

    public void handleWrite(SelectionKey key) throws IOException {

       // 附加到 SelectionKey 上的 ByteBuffer 包含了之前从信道中读取的数据。

       ByteBuffer buf = (ByteBuffer)key.attachment();

       // 该方法用来修改缓冲区的内部状态，以指示 write 操作从什么地方获取数据，及还剩多少数据

       buf.flip();

       SocketChannel channel = (SocketChannel)key.channel(); // 获取信道

       channel.write(buf); // 向信道中写数据

       if (!buf.hasRemaining()){

           // 如果没有剩余数据可读，则修改该键关联的操作集，指示其只能进行读操作了

           key.interestOps(SelectionKey. OP_READ );

       }

    // 如果缓冲区中还有剩余数据，该操作将剩余数据移到缓冲区前端，以使下次迭代能读入更多数据。

       buf.compact();

    }

}

下面 是回显服务器端代码的实现，在服务器端创建一个选择器，并将其与每个侦听客户端连接的套接字说对应的 ServerSocketChannel 注册在一起。然后进行反复循环，调用 select() 方法，并调用相应的操作器对各种类型的 I/O 操作进行处理 。

package org.zapldy.tcpip.nio;

 

import java.io.IOException;

import java.net.InetSocketAddress;

import java.nio.channels.SelectionKey;

import java.nio.channels.Selector;

import java.nio.channels.ServerSocketChannel;

import java.util.Iterator;

 

public class EchoSelectorServer {

    private static final int BUFSIZE = 256;

    private static final int TIMEOUT = 3000;

   

    private static final int PORT = 8888;

   

    public static void main(String[] args) throws IOException{

       Selector selector = Selector.open ();

      

       ServerSocketChannel listnChannel = ServerSocketChannel.open ();

       listnChannel.socket().bind( new InetSocketAddress( PORT ));

       // 只有非阻塞信道才可以注册选择器，因此需要将其配置为适当的状态

       listnChannel.configureBlocking( false );

       // 在注册过程中指出该信道可以进行 “accept” 操作

       listnChannel.register(selector, SelectionKey. OP_ACCEPT );

      

        Protocol protocol = new EchoProtocol( BUFSIZE );

       while ( true ){

           if (selector.select( TIMEOUT ) == 0){

              System. out .print( "==" );

              continue ;

           }

          

                          Iterator<SelectionKey> keyIter =

selector.selectedKeys().iterator();

           while (keyIter.hasNext()){

              SelectionKey key = keyIter.next();

              if (key.isAcceptable()){

                  protocol.handleAccept(key);

              }

              if (key.isReadable()){

                  protocol.handleRead(key);

              }

              if (key.isWritable() && key.isValid()){

                  protocol.handleWrite(key);

              }

              // 由于 select() 操作只是向 Selector 所关联的键集合中添加元素

                // 因此，如果不移除每个处理过的键，

// 它就会在下次调用 select() 方法时仍然保留在集合中

                // 而且可能会有无用的操作来调用它。

              keyIter.remove();

           }

       }

    }

}

4、 流（ TCP ）信道详解

流信道有两个变体： SocketChannel 和 ServerSocketChannel 。像其对应的 Socket 一样，SocketChannel 是相互连接的终端进行通信的信道。

l  SocketChannel ：创建，连接和关闭

    static SocketChannel open(SocketAddress remote)

    static SocketChannel open()

    boolean connect(SocketAddress remote)

    boolean isConnected()

    void close()

    boolean isOpen()

    Socket socket()

    调用SocketChannel 的静态工厂方法open() 可以创建一个实例。open() 方法的第一种形式以SocketAddress 为参数，返回一个连接到指定服务器的SocketChannel 实例。注意，该方法可能会无限期地阻塞下去。open() 的无参数形式用于创建一个没有连接的SocketChannel 实例，该实例可以通过调用connect() 方法连接到指定终端。当使用完SocketChannel 后，需要调用close() 方法将其关闭。有一点很重要，即每个SocketChannel 实例都包裹了一个基本的Java Socket, 并可以通过socket() 方法对该Socket 进行访问。这就可以通过基本的Socket 方法进行绑定、设置套接字选项等操作。

         在创建并连接 SocketChannel 后，就可以调用该信道的读写方法进行 I/O 操作。

SocketChannel ： 读和写

    int read(ByteBuffer dst)

    long read(ByteBuffer[] dsts)

    long read(ByteBuffer[] dsts, int offset, int length)

    int write(ByteBuffer src)

    int write(ByteBuffer[] srcs)

    int write(ByteBuffer[] srcs, int offset, int length)

         读操作的最基本形式以一个ByteBuffer 为参数，并将读取的数据填入该缓冲区所有剩余字节空间中。另一种形式以多个ByteBuffer 为参数（ByteBuffer 数组），并根据其在数组中的顺序，将读取的数据依次填入每个缓冲区的剩余字节空间中。这种方法称为散射式读，因为它将读入的直接分散到了多个缓冲区中。需要注意重要的一点，散射式读不一定会将所有缓冲区填满，这些缓冲区的总空间大小只是一个上限。

       写操作的最基本形式是以一个ByteBuffer 为参数，并试图将该缓冲区中剩余的字节写入信道。另一种形式以一个ByteBuffer 数组作为参数，并试图将所有缓冲区中的剩余字节都写入信道。这种方法称为聚集式写，因为它把多个缓冲区中的字节聚集起来，一起发送出去。

         与其对应的 ServerSocket 一样， ServerSocketChannel 是用来侦听客户端连接的信道。

ServerSocketChannel ： 创建，接受和关闭

    static ServerSocketChannel open()

    ServerSocket socket()

    SocketChannel accept()

    void close()

    boolean isOpen()

    调用静态工厂方法open() 可以创建一个ServerSocketChannel 实例。每个实例都包裹了一个ServerSocket 实例，并可以通过socket() 方法对其访问。正如前面的例子所表明的，必须通过底层的ServerSocket 实例来实现绑定制定端口，设置套接字选项等操作。在创建了信道实例并绑定端口后，就可以调用accept() 方法来准备接收客户端的连接请求。连接成功则返回一个新的已连接的SocketChannel 。在用完ServerSocketChannel 后，需要调用close() 方法将其关闭。

    如前文提到的那样，阻塞式信道除了能够（必须）与 Buffer 一起使用外，对于普通套接字来说几乎没有优点。因此，可能总是需要将其设置成非阻塞式的。

SocketChannel, ServerSocketChannel ： 设置阻塞行为

    SelectableChannel configureBlocking(boolean block)

    boolean isBlocking()

    通过调用configureBlocking(false) 可以将SocketChannel 或ServerSocketChannel 设置为非阻塞模式。configureBlocking() 方法将返回一个SelectableChannel ，它是SocketChannel 和ServerSocketChannel 父类。

    考虑为 SocketChannel 设置连接的情况。如果传给 SocketChannel 的工厂方法 open() 一个远程地址，对该方法的调用则将阻塞等待，直到成功建立了连接。要避免这种情况，可以使用 open() 方法的无参数形式，配置信道为非阻塞模式，再调用 connect() 方法，制定远程终端地址。如果在没有阻塞的情况下连接已经建立， connect() 方法返回 true ；否则需要有检查套接字是否连接成功的方法。

SocketChannel ： 测试连接性

    boolean finishConnect()

    boolean isConnected()

    boolean isConnectionPending()

    对于非阻塞SocketChannel 来说，一旦已经发起连接，底层套接字可能既不是已经连接，又不是没有连接，而是连接“正在进行”。由于底层协议的工作机制，套接字可能会在这个状态一直保持下去。finishConnect() 方法可以用来检查在非阻塞套接字上试图进行的连接状态，还可以在阻塞套接字建立连接的过程中阻塞等待，直到连接成功建立。例如，你可能需要将信道配置成非阻塞模式，通过connect() 方法发起连接，做完一些其他工作后，又将信道配置成阻塞模式，然后调用finishConnect() 方法等待连接建立完成。或者可以让信道保持在非阻塞模式，并反复调用finishConnect() 方法。如TCPEchoClientNoblocking 类中所示。

       isConnected() 用于检查套接字是否已经建立了连接，从而避免在进行其他操作时抛出NotYetConnectedException 异常（如在调用read() 或write() 时）。还可以使用isConnectedPending() 方法来检查是否有连接在该信道上发起。知道是否有连接发起是有必要的，因为如果没有的话，finishConnect() 方法将抛出NoConnectionPendingException 异常。

5、 Selector 详解

EchoSelectorServer 示例中展示了 Selector 的基本用法。在此，我们将对其进行更加详细的介绍。

Selector ： 创建和关闭

    static Selector open()

    boolean isOpen()

    void close()

    调用Selector 的open() 工厂方法可以创建一个选择器实例。选择器的状态是“打开”或是“关闭”的。创建时选择器的状态时打开的，并保持该状态，直到调用close() 方法通知系统其任务已经完成。可以调用isOpen() 方法来检查选择器是否已经关闭。

1) 在信道中注册

    我们已经知道，每个选择器都有一组与之关联的信道，选择器对这些信道上“感兴趣的” I/O 操作进行监听。 Selector 与 Channel 之间的关联由一个 SelectionKey 实例表示。（注意：一个信道可以注册多个Selector 实例，因此可以有多个关联的 SelectionKey 实例）。 SelectionKey 维护了一个信道上感兴趣的操作类型信息，并将这些信息存放在一个 int 型的位图中，该 int 型数据的每一位都有相应的含义。

    SelectionKey 类中的常量定义了信道上可能感兴趣的操作类型，每个这种常量都是只有一位设置为 1 的位掩码。

SelectionKey ： 兴趣操作集

    static int OP_ACCEPT

    static int OP_CONNECT

    static int OP_READ

    static int OP_WRITE

    int interestOps()

    SelectionKey interestOps( int ops)

    通过对OP_ACCEPT ，OP_CONNECT ，OP_READ 以及OP_WRITE 中适当的常量进行按位OR ，我们可以构造一个位向量来制定一组操作。例如，一个包含读和写的操作集可由表达式（OP_READ | OP_WRITE ）来指定。不带参数的interestOps() 方法将返回一个int 型位图，该位图中设置为1 的每一位都指示了信道上需要监听的一种操作。另一种方法以一个位图为参数，指示了应该监听信道上的哪些操作。重点提示：任何对key （信道）所关联的兴趣操作集的改变，都只在下次调用了select() 方法后才会生效。

 

SocketChannel, ServerSocketChannel ： 注册 Selector

    SelectionKey register(Selector sel, int ops)

    SelectionKey register(Selector sel, int ops, Object attachment)

    int validOps()

    boolean isRegistered()

    SelectionKey keyFor(Selector sel)

    调用信道的register() 方法可以将一个选择器注册到该信道。在注册过程中，通过存储在int 型数据中的位图来指定该信道上的初始兴趣操作集。register() 方法将返回一个代表了信道和给定选择器之间的关联的SelectionKey 实例。validOps() 方法用于返回一个指示了该信道上的有效I/O 操作集的位图。对于SocketChannel 来说，有效操作包括读、写和连接。一个信道可能只与一个选择器注册一次，因此后续对register() 方法的调用只是简单地更新该key所关联的兴趣操作集。使用isRegistered() 方法可以检查信道是否已经注册了选择器。keyFor() 方法与第一次调用register() 方法返回的是同一个SelectionKey 实例，除非该信道没有注册给定的选择器。

以下代码注册了一个信道，支持读写操作：

SelectionKey key = clientChannel.register(selector,

                          SelectionKey.OP_READ | SelectionKey.OP_WRITE)

下图展示了一个选择器，其键集中包含了 7 个代表注册信道的键：两个在端口 8888 和 8889 上的服务器信道，以及从服务器信道创建的 5 个客户端信道：

SelectionKey ： 获取和取消

    Selector selector()

    SelectableChannel channel()

    void cancel()

    键关联的 Selector 实例和 Channel 实例可以分别使用该键的 selector() 和 channel() 方法获得。cancel() 方法用于（永久性地）注销该键，并将其放入选择器的注销集中。在下一次调用 select() 方法时，这些键将从该选择器的所有集中移除，其关联的信道也将不再被监听（除非它又重新注册）。

 

2) 选取和识别准备就绪的信道

    在信道上注册了选择器，并由关联的键指定了感兴趣的 I/O 操作集后，我们就只需要坐下来等待 I/O 了。这要使用选择器来完成。

Selector ： 等待信道准备就绪

    int select()

    int select(long timeout)

    int selectNow()

    Selector wakeup()

    select() 方法用于从已经注册的信道中返回在感兴趣的I/O 操作集上准备就绪的信道总数。（例如，兴趣操作集中包含OP_READ 的信道有数据可读，或包含OP_ACCEPT 的信道有连接请求待接受。）以上三个select() 方法的唯一区别在于它们的阻塞行为。无参数的select() 方法会阻塞等待，直到至少有一个注册信道中有感兴趣的操作准备就绪，或有别的线程调用了该选择器wakeup() 方法（这种情况下select() 方法将返回0 ）。以超时时长作为参数的select 方法也会阻塞等待，直到至少有一个信道准备就绪，或等待时间超过了指定的毫秒数（正数），或者有另一个线程调用其wakeup() 方法。selectNow() 方法是一个非阻塞版本：它总数立即返回，如果没有信道准备就绪，则返回0.wakeup() 方法可以使用当前阻塞（也就是说在另一个线程中阻塞）的任何一种select() 方法立即返回；如果当前没有select 方法阻塞，下一次调用者三种方法的任何一个都将立即返回。

       选择之后，我们需要知道哪些信道准备好了特定的 I/O 操作。每个选择器都维护了一个已选键集，与这些键关联的信道都有即将发生的特定 I/O 操作。通过调用 selectedKey() 方法可以访问已选键集，该方法返回一组 selectionKey 。我们可以在这组键上进行迭代，分别处理等待在每个键关联的信道上的 I/O 操作。

    Iterator<SelectionKey> keyIter =

selector.selectedKeys().iterator();

while (keyIter.hasNext()){

         SelectionKey key = keyIter.next();

         //... 在这里处理该 key 所关联的信道 channel

         keyIter.remove();

       }

    }

 

Selector ： 获取键集

    Set<SelectionKey> keys()

    Set<SelectionKey> selectedKeys()

       以上方法返回选择器的不同键集。keys() 方法返回当前已注册的所有键。返回的键集是不可修改的；任何对其进行修改的尝试（如，调用其remove() 方法）都将抛出UnsupportedOperationException 异常。selectedKeys() 方法用于返回上次调用select() 方法时，被“选中”的已准备好进行I/O 操作的键。重要提示：selectedKeys() 方法返回的键是可修改的，在实际上在两次调用select() 方法之间，都必须“手工”将清空。换句话说，select 方法只会在已有的所选键集上添加键，它们不会创建新的键集。

    所有键集指示了哪些信道当前可以进行 I/O 操作。对于选中的每个信道，我们需要知道它们各自准备好的特定 I/O 操作。除了兴趣操作集外，每个键还维护了一个即将进行的 I/O 操作集，称为就绪操作集。

SelectionKey ： 查找就绪的I/O 操作

    int readyOps()

    boolean isAcceptable()

    boolean isConnectable()

    boolean isReadable()

    boolean isValid()

    boolean isWritable()

    对于给定的键，可以使用readyOps() 方法或其他指示方法来确定兴趣集中的哪些I/O 操作可以执行。readyOps() 方法以位图的形式返回所有准备就绪的操作集。其他方法用于分别检查各种操作是否可用。

例如，查看键关联的信道上是否有正在等待的读操作，可以使用以下代码：

(key.readOps() & SelectionKey.OP_READ) != 0

或

key.isReadable()

    选择器的已选键集中的键，以及每个键中准备就绪的操作，都是由 select() 方法来确定的。随着时间的推进，这些信息可能会过时。其他线程可能会处理准备就绪的 I/O 操作。同时，键也不是永远存在的。但其关联的信道或选择器关闭时，键也将失效。通过调用其 cancel() 方法可以显示地将键设置为无效。调用其 isValid()方法可以检测一个键的有效性。无效的键将添加到选择器的注销集中，并在下次调用任意一种形式的 select()方法和或者 close() 方法时从键集中移除。（当然，从键集中移除键意味着与它关联的信道也不再受监听。）

 

3) 信道附件

    当一个信道准备好进行 I/O 操作时，通常还需要额外的信息来处理请求。例如，在前面的回显协议中，但客户端信道准备好写操作时，就需要有数据可写。当然，我们所需要的可写数据是由之前同一信道上的读操作收集的，但是在其可写之前，这些数据存放在什么地方呢？另一个例子，如果一个消息一次传来了多个字节，我们需要保存已接收的部分消息，直到整个消息接收完成。这两种情况都需要维护每个信道的状态信息。然而，我们非常幸运！ SelectionKey 通过使用附件使保存每个信道的状态变得容易。

SelectionKey ： 查找就绪的I/O 操作

Object attach(Object ob)

Object attachment()

         每个键可以有一个附件，数据类型只能是Object 类。附件可以在信道第一次调用register() 方法时与之关联，或者后来再使用attach() 方法直接添加到键上。通过SelectionKey 的attachment() 方法可以访问键的附件。

 

4) Selector 小结

    总的来说，使用 Selector 的步骤如下：

1、 创建一个 Selector 实例。

2、 将其注册到各种信道，指定每个信道上感兴趣的 I/O 操作。

3、 重复执行：

1） 调用一种 select 方法

2） 获取选取的键列表

3） 对于已选键集中的每个键。

a.  获取信道，并从键中获取附件（如果合适的话）

b.  确定准备就绪的操作并执行。如果是 accept 操作，将接受的信道设置为非阻塞模式，并将其与选择器注册。

c.  如果需要，修改键的兴趣操作集

d.  从已选键中移除键

如果选择器告诉了你什么时候 I/O 操作准备就绪，你还需要非阻塞 I/O 吗？答案是肯定的。信道在已选键集中的键并不能确保非阻塞 I/O ，因为调用了 select() 方法后，键集信息可能会过时。另外，阻塞式写操作会阻塞等待直到写完所有字节，而就绪集中的 OP_WRITE 仅表示至少有一个字节可写。实际上，只是非阻塞模式的信道才能与选择器进行注册：如果信道在阻塞模式， SelectableChannel 类的 register() 方法将抛出 IllegalBlockingModeException 异常。

 

后记：

这里只是对 NIO 做了一个基础的讲解，我将会在后续的文章中完善更多关于 NIO 底层实现的内容。