Java NIO 原理图文分析及代码实现

前言:

最近在分析 hadoop 的 RPC(Remote Procedure Call Protocol ，远程过程调用协议，它是一种通过网络从远程计算机程序上请求服务，而不需要了解底层网络技术的协议。可以参考：http://baike.baidu.com/view/32726.htm ）机制时，发现 hadoop 的 RPC 机制的实现主要用到了两个技术：动态代理（动态代理可以参考博客：http://weixiaolu.iteye.com/blog/1477774 ）和 java NIO。为了能够正确地分析 hadoop 的 RPC 源码，我觉得很有必要先研究一下 java NIO 的原理和具体实现。

这篇博客我主要从两个方向来分析 java NIO

一．java NIO 和阻塞 I/O 的区别

 1\. 阻塞I/O通信模型

 2\. java NIO原理及通信模型

二．java NIO 服务端和客户端代码实现

具体分析：

一．java NIO 和阻塞 I/O 的区别

1. 阻塞 I/O 通信模型

假如现在你对阻塞 I/O 已有了一定了解，我们知道阻塞 I/O 在调用 InputStream.read()方法时是阻塞的，它会一直等到数据到来时（或超时）才会返回；同样，在调用 ServerSocket.accept()方法时，也会一直阻塞到有客户端连接才会返回，每个客户端连接过来后，服务端都会启动一个线程去处理该客户端的请求。阻塞 I/O 的通信模型示意图如下：

如果你细细分析，一定会发现阻塞 I/O 存在一些缺点。根据阻塞 I/O 通信模型，我总结了它的两点缺点：

1. 当客户端多时，会创建大量的处理线程。且每个线程都要占用栈空间和一些 CPU 时间

2. 阻塞可能带来频繁的上下文切换，且大部分上下文切换可能是无意义的。

在这种情况下非阻塞式 I/O 就有了它的应用前景。

java NIO 原理及通信模型

Java NIO 是在 jdk1.4 开始使用的，它既可以说成“新 I/O”，也可以说成非阻塞式 I/O。下面是 java NIO 的工作原理：

1. 由一个专门的线程来处理所有的 IO 事件，并负责分发。

2. 事件驱动机制：事件到的时候触发，而不是同步的去监视事件。

3. 线程通讯：线程之间通过 wait,notify 等方式通讯。保证每次上下文切换都是有意义的。减少无谓的线程切换。

阅读过一些资料之后，下面贴出我理解的 java NIO 的工作原理图：

（注：每个线程的处理流程大概都是读取数据、解码、计算处理、编码、发送响应。）

Java NIO 的服务端只需启动一个专门的线程来处理所有的 IO 事件，这种通信模型是怎么实现的呢？呵呵，我们一起来探究它的奥秘吧。java NIO 采用了双向通道（channel）进行数据传输，而不是单向的流（stream），在通道上可以注册我们感兴趣的事件。一共有以下四种事件：

服务端和客户端各自维护一个管理通道的对象，我们称之为 selector，该对象能检测一个或多个通道 (channel) 上的事件。我们以服务端为例，如果服务端的 selector 上注册了读事件，某时刻客户端给服务端发送了一些数据，阻塞 I/O 这时会调用 read()方法阻塞地读取数据，而 NIO 的服务端会在 selector 中添加一个读事件。服务端的处理线程会轮询地访问 selector，如果访问 selector 时发现有感兴趣的事件到达，则处理这些事件，如果没有感兴趣的事件到达，则处理线程会一直阻塞直到感兴趣的事件到达为止。下面是我理解的 java NIO 的通信模型示意图：

二．java NIO 服务端和客户端代码实现

为了更好地理解 java NIO,下面贴出服务端和客户端的简单代码实现。

服务端：

Java 代码

package cn.nio;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
/**
- NIO 服务端
- @author 小路
*/
public class NIOServer {
```
 //通道管理器  
```
```
 private Selector selector;  
```
```
 /** 
```

  * 获得一个ServerSocket通道，并对该通道做一些初始化的工作

```
  * @param port  绑定的端口号 
```
```
  * @throws IOException 
```
```
  */  
```

 public void initServer(int port) throws IOException {

     // 获得一个ServerSocket通道

     ServerSocketChannel serverChannel = ServerSocketChannel.open();

```
     // 设置通道为非阻塞  
```

     serverChannel.configureBlocking(false);

     // 将该通道对应的ServerSocket绑定到port端口

     serverChannel.socket().bind(new InetSocketAddress(port));

```
     // 获得一个通道管理器  
```

     this.selector = Selector.open();

     //将通道管理器和该通道绑定，并为该通道注册SelectionKey.OP_ACCEPT事件,注册该事件后，

     //当该事件到达时，selector.select()会返回，如果该事件没到达selector.select()会一直阻塞。

     serverChannel.register(selector, SelectionKey.OP_ACCEPT);

```
 }  
```
```
 /** 
```

  * 采用轮询的方式监听selector上是否有需要处理的事件，如果有，则进行处理

```
  * @throws IOException 
```
```
  */  
```
```
 @SuppressWarnings("unchecked")  
```

 public void listen() throws IOException {

     System.out.println("服务端启动成功！");

```
     // 轮询访问selector  
```
```
     while (true) {  
```

         //当注册的事件到达时，方法返回；否则,该方法会一直阻塞

```
         selector.select();  
```

         // 获得selector中选中的项的迭代器，选中的项为注册的事件

         Iterator ite = this.selector.selectedKeys().iterator();

```
         while (ite.hasNext()) {  
```

             SelectionKey key = (SelectionKey) ite.next();

             // 删除已选的key,以防重复处理

```
             ite.remove();  
```

             // 客户端请求连接事件

             if (key.isAcceptable()) {

                 ServerSocketChannel server = (ServerSocketChannel) key

```
                         .channel();  
```

                 // 获得和客户端连接的通道

                 SocketChannel channel = server.accept();

                 // 设置成非阻塞

                 channel.configureBlocking(false);

                 //在这里可以给客户端发送信息哦

                 channel.write(ByteBuffer.wrap(new String("向客户端发送了一条信息").getBytes()));

                 //在和客户端连接成功之后，为了可以接收到客户端的信息，需要给通道设置读的权限。

                 channel.register(this.selector, SelectionKey.OP_READ);

                 // 获得了可读的事件

             } else if (key.isReadable()) {

```
                     read(key);  
```
```
             }  
```
```
         }  
```
```
     }  
```
```
 }  
```
```
 /** 
```

  * 处理读取客户端发来的信息 的事件

```
  * @param key 
```
```
  * @throws IOException  
```
```
  */  
```

 public void read(SelectionKey key) throws IOException{

     // 服务器可读取消息:得到事件发生的Socket通道

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

```
     // 创建读取的缓冲区  
```

     ByteBuffer buffer = ByteBuffer.allocate(10);

```
     channel.read(buffer);  
```
```
     byte[] data = buffer.array();  
```

     String msg = new String(data).trim();

     System.out.println("服务端收到信息："+msg);

     ByteBuffer outBuffer = ByteBuffer.wrap(msg.getBytes());

     channel.write(outBuffer);// 将消息回送给客户端

```
 }  
```
```
 /** 
```
```
  * 启动服务端测试 
```
```
  * @throws IOException  
```
```
  */  
```

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

     NIOServer server = new NIOServer();

```
     server.initServer(8000);  
```
```
     server.listen();  
```
```
 }  
```
}

客户端：

Java 代码

package cn.nio;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
/**
- NIO 客户端
- @author 小路
*/
public class NIOClient {
```
 //通道管理器  
```
```
 private Selector selector;  
```
```
 /** 
```

  * 获得一个Socket通道，并对该通道做一些初始化的工作

```
  * @param ip 连接的服务器的ip 
```

  * @param port  连接的服务器的端口号

```
  * @throws IOException 
```
```
  */  
```

 public void initClient(String ip,int port) throws IOException {

```
     // 获得一个Socket通道  
```

     SocketChannel channel = SocketChannel.open();

```
     // 设置通道为非阻塞  
```

     channel.configureBlocking(false);

```
     // 获得一个通道管理器  
```

     this.selector = Selector.open();

     // 客户端连接服务器,其实方法执行并没有实现连接，需要在listen（）方法中调

     //用channel.finishConnect();才能完成连接

     channel.connect(new InetSocketAddress(ip,port));

     //将通道管理器和该通道绑定，并为该通道注册SelectionKey.OP_CONNECT事件。

     channel.register(selector, SelectionKey.OP_CONNECT);

```
 }  
```
```
 /** 
```

  * 采用轮询的方式监听selector上是否有需要处理的事件，如果有，则进行处理

```
  * @throws IOException 
```
```
  */  
```
```
 @SuppressWarnings("unchecked")  
```

 public void listen() throws IOException {

```
     // 轮询访问selector  
```
```
     while (true) {  
```
```
         selector.select();  
```

         // 获得selector中选中的项的迭代器

         Iterator ite = this.selector.selectedKeys().iterator();

```
         while (ite.hasNext()) {  
```

             SelectionKey key = (SelectionKey) ite.next();

             // 删除已选的key,以防重复处理

```
             ite.remove();  
```
```
             // 连接事件发生  
```

             if (key.isConnectable()) {

                 SocketChannel channel = (SocketChannel) key

```
                         .channel();  
```

                 // 如果正在连接，则完成连接

                 if(channel.isConnectionPending()){

                     channel.finishConnect();

```
                 }  
```

                 // 设置成非阻塞

                 channel.configureBlocking(false);

                 //在这里可以给服务端发送信息哦

                 channel.write(ByteBuffer.wrap(new String("向服务端发送了一条信息").getBytes()));

                 //在和服务端连接成功之后，为了可以接收到服务端的信息，需要给通道设置读的权限。

                 channel.register(this.selector, SelectionKey.OP_READ);

                 // 获得了可读的事件

             } else if (key.isReadable()) {

```
                     read(key);  
```
```
             }  
```
```
         }  
```
```
     }  
```
```
 }  
```
```
 /** 
```

  * 处理读取服务端发来的信息 的事件

```
  * @param key 
```
```
  * @throws IOException  
```
```
  */  
```

 public void read(SelectionKey key) throws IOException{

     //和服务端的read方法一样

```
 }  
```
```
 /** 
```
```
  * 启动客户端测试 
```
```
  * @throws IOException  
```
```
  */  
```

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

     NIOClient client = new NIOClient();

     client.initClient("localhost",8000);

```
     client.listen();  
```
```
 }  
```
}

小结：

终于把动态代理和 java NIO 分析完了，呵呵，下面就要分析 hadoop 的 RPC 机制源码了，博客地址：http://weixiaolu.iteye.com/blog/1504898 。不过如果对 java NIO 的理解存在异议的，欢迎一起讨论。

Java 项目发布到 Maven 中央仓库小记

[图片] 我 20 年就已经注册并成功发布过一次项目因此略过注册环节 1.填写项目相关信息 [图片] 2.等待 bot 提示你验证域名所有权。如果没有域名可以使用 io.github.xxx 作为 groupId 切记不要填成 com.github [图片] 3.去你所在的域名服务商加一个 TXT 记录 OSSRH-9 ..

软件架构设计原则

1.1 开闭原则开闭原则（Open-Closed Principle，COP）是指一个软件实体（如类、模块和函数）应该对扩展开放，对修改关闭。所谓的关闭，也正是对扩张和修改两个行为的一个原则。它强调的是用抽象构建框架，用实现扩展细节，可以提高软件系统的客服用心及可维护性。开闭原则是对面向对象设计最基础的设计原则，它知 ..

别以为“自动挡”就不可能出现 OOM

这里的“自动挡”，是我对 Java 自动垃圾收集器的戏称。的确，经过这么多年的发展， Java 的垃圾收集器已经非常成熟了。有了自动垃圾收集器，绝大多数情况下我们写程序时可以专注于业务逻辑，无需过多考虑对象的分配和释放，一般也不会出现 OOM。　　但，内存空间始终是有限的，Java 的几大内存区域始终都有 OOM 的 ..

用好 Java 8 的日期时间类，少踩一些“老三样”的

在 Java 8 之前，我们处理日期时间需求时，使用 Date、Calender 和 SimpleDateFormat，来声明时间戳、使用日历处理日期和格式化解析日期时间。但是，这些类的 API 的缺点比较明显，比如可读性差、易用性差、使用起来冗余繁琐，还有线程安全问题。　　因此，Java 8 推出了新的日期时间类。 ..

Java 入门

.java-> 编译->.class(字节码文件)-> 解释-> 机器码字节码解释成机器码是实时进行的，从而导致每次执行时都需要解释，这也是 java 性能不如 c/c++ 的原因之一。这样做是为了实现跨平台即时编译（JIT, just-in-time）：将解释出来的机器码保存到内存中，再次 ..

为什么选择 GraalVM？

GraalVM 是一款强大的虚拟机，它为 Java 开发者提供了许多独特的优势和功能。无论是在资源利用率、启动速度、安全性还是与其他语言的集成方面，GraalVM 都是一个非常有吸引力的选择。以下是选择 GraalVM 的几个主要原因：低资源使用 GraalVM 通过提前编译 Java 应用程序为独立的二进制文件， ..

欢迎来到这里！

我们正在构建一个小众社区，大家在这里相互信任，以平等 • 自由 • 奔放的价值观进行分享交流。最终，希望大家能够找到与自己志同道合的伙伴，共同成长。

关于

Java NIO 原理图文分析及代码实现

相关帖子

Java 项目发布到 Maven 中央仓库小记

软件架构设计原则

别以为“自动挡”就不可能出现 OOM

用好 Java 8 的日期时间类，少踩一些“老三样”的

隐写分析使用方法

Java 入门

为什么选择 GraalVM？

欢迎来到这里！

近期热议

推荐标签标签

最新标签

Java NIO 原理图文分析及代码实现

相关帖子

Java 项目发布到 Maven 中央仓库小记

软件架构设计原则

别以为“自动挡”就不可能出现 OOM

用好 Java 8 的日期时间类，少踩一些“老三样”的

隐写分析使用方法

Java 入门

为什么选择 GraalVM？

欢迎来到这里！

近期热议

推荐标签 标签

最新标签

推荐标签标签