Java NIO的基本概念Channel、Buffer、Selector以及非阻塞网络通信案例

深藏阁楼爱情的钟 2022-09-11 05:20 75阅读 0赞

> 详细介绍了Java NIO中的基本概念，Buffer、Channel、Selector，以及NIO非阻塞网络通信的基本案例。

### 文章目录 ###

*  1 基本概念
 *   *   *  1.1 同步和异步
         *  1.2 阻塞和非阻塞
 *  2 通道（Channel）与缓冲区（Buffer）
 *   *  2.1 缓冲区（Buffer）
     *   *  2.1.1 缓冲区的基本属性
         *  2.1.2 缓冲区的基本操作与常用方法
         *  2.1.3 缓冲区的数据操作
         *  2.1.4 直接与非直接缓冲区
     *  2.2 通道（Channel）
     *   *  2.2.1 Channel的实现
         *   *  2.2.2 获取通道
 *  3 选择器Selector
 *   *  3.1 创建Selector
     *  3.2 向选择器注册通道
     *  3.3 监听事件
     *  3.4 Selector 的常用方法
 *  4 NIO非阻塞网络通信案例
 *   *  4.1 非阻塞式TCP通信
     *  4.2 非阻塞式UDP通信

# 1 基本概念 #

### 1.1 同步和异步 ###

1.  同步I/O:每个请求必须逐个地被处理，一个请求的处理会导致整个流程的暂时等待，这些事件无法并发地执行。用户线程发起I/O请求后需要等待或者轮询内核I/O操作完成后才能继续执行。
2.  异步I/O:多个请求可以并发地执行，一个请求或者任务的执行不会导致整个流程的暂时等待。用户线程发起I/O请求后仍然继续执行，当内核I/O操作完成后会通知用户线程，或者调用用户线程注册的回调函数。

**传统IO和NIO都是同步的，AIO是异步的。**

### 1.2 阻塞和非阻塞 ###

阻塞和非阻塞是进程在访问数据的时候，请求操作是否准备就绪的一种处理方式。

1.  阻塞:某个请求发出后，由于该请求操作需要的条件不满足，请求操作一直阻塞，不会返回，直到条件满足。
2.  非阻塞:请求发出后，若该请求需要的条件不满足，则立即返回一个标志信息告知条件不满足，而不会一直等待。一般需要通过循环判断请求条件是否满足来获取请求结果。

**同步和异步着重点在于多个任务执行过程中，后发起的任务是否必须等先发起的任务完成之后再进行。不管先发起的任务请求是阻塞等待完成，还是立即返回通过循环等待请求成功。而阻塞和非阻塞重点在于请求的方法是否在条件不满足时被阻塞，是否立即返回。**

**传统的IO 流都是阻塞式的**。也就是说，当一个线程调用read() 或write() 时，该线程被阻塞，直到有一些数据被读取或写入，该线程在此期间不能执行其他任务。因此，在完成网络通信进行IO 操作时，由于线程会阻塞，所以服务器端必须为每个客户端都提供一个独立的线程进行处理，即多线程方案,但是当服务器端需要处理大量客户端时，仍然会造成大量线程等待,性能仍然急剧下降。

**Java NIO 是非阻塞模式的**。当线程从某通道进行读写数据时，若没有数据可用时，该线程可以进行其他任务。线程通常将非阻塞IO 的空闲时间用于在其他通道上执行IO 操作，所以单独的线程可以管理多个输入和输出通道。因此，NIO 可以让服务器端使用一个或有限几个线程来同时处理连接到服务器端的所有客户端。

# 2 通道（Channel）与缓冲区（Buffer） #

**Java NIO系统的核心在于：通道(Channel)和缓冲区(Buffer)。通道表示打开到IO 设备(例如：文件、套接字)的连接。若需要使用NIO 系统，需要获取用于连接IO 设备的通道以及用于容纳数据的缓冲区。然后操作缓冲区，对数据进行处理。**

**简而言之，Channel 负责传输，Buffer 负责存储。**

## 2.1 缓冲区（Buffer） ##

一个用于特定基本数据类型的容器。由java.nio 包定义的，所有缓冲区都是Buffer 抽象类的子类。Java NIO 中的Buffer 主要用于与NIO 通道进行交互。在读取数据时，它是直接读到缓冲区中的; 在写入数据时，写入到缓冲区中。任何时候访问NIO中的数据，都是通过缓冲区进行操作。

Buffer就像一个数组，可以保存多个相同类型的数据。但它不仅仅是一个数组。缓冲区提供了对数据的结构化访问，而且还可以跟踪系统的读/写进程。根据数据类型不同(boolean 除外)，有以下Buffer 常用子类：

**ByteBuffer、CharBuffer、ShortBuffer、IntBuffer、LongBuffer、FloatBuffer、DoubleBuffer。**

上述Buffer 类他们都采用相似的方法进行管理数据，只是各自管理的数据类型不同而已。都是通过对应的allocate(int capacity)方法获取一个Buffer 对象，该方法表示创建一个缓冲区容量为capacity 的XxxBuffer对象！

### 2.1.1 缓冲区的基本属性 ###

public abstract class Buffer {
        
        private int mark = -1;
        private int position = 0;
        private int limit;
        private int capacity;
       .............
    ｝

**容量(capacity)** ：表示Buffer 最大数据容量，缓冲区容量不能为负，并且创建后不能更改。

**限制/界限(limit)**：第一个不可读取或写入的数据的索引，即位于limit索引以及之后的数据不可读写。缓冲区的限制不能为负，并且不能大于其容量。写模式下，limit等于Buffer的capacity。读模式时，limit表示你最多能读到多少数据，即limit会被设置成写模式下的position值。

**位置(position)**：将要读取或写入的数据的索引。缓冲区的位置不能为负，并且不能大于其限制。最小为0，最大可为capacity – 1.

**标记(mark)与重置(reset)**：标记是一个索引，通过Buffer中的mark()方法指定Buffer中一个特定的（当前的）position，之后可以通过调用reset()方法恢复到这个position.mark应该小于等于position和limit，如果调整了位置之后mark大于这两数的位置的话，maark将被置为-1.

> \-1 <= mark <= position <= limit <= capacity

### 2.1.2 缓冲区的基本操作与常用方法 ###

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-S0jOdIl3-1631502083167)(C:\\Users\\lx\\AppData\\Roaming\\Typora\\typora-user-images\\image-20210913105354392.png)\]

Buffer 的常用方法：

<table> 
 <thead> 
  <tr> 
   <th>方法</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Buffer clear()</td> 
   <td>清空缓冲区并返回对缓冲区的引用。但是缓冲区的数据还在，只是处于被遗忘的状态，即指针初始化。可以通过get()验证，还能得到第一个数据。或者hasRemaining验证</td> 
  </tr> 
  <tr> 
   <td>Buffer flip()</td> 
   <td>将缓冲区的限制设置为当前位置所在的索引，并将当前位置的索引重置为0。切换到读模式。</td> 
  </tr> 
  <tr> 
   <td>int capacity()</td> 
   <td>返回Buffer 的capacity大小</td> 
  </tr> 
  <tr> 
   <td>boolean hasRemaining()</td> 
   <td>判断缓冲区中是否还有元素</td> 
  </tr> 
  <tr> 
   <td>int limit()</td> 
   <td>返回Buffer 的限制(limit)</td> 
  </tr> 
  <tr> 
   <td>Bufferlimit(int n)</td> 
   <td>将设置缓冲区界限为n,并返回一个具有新limit 的缓冲区对象</td> 
  </tr> 
  <tr> 
   <td>Buffer mark()</td> 
   <td>对缓冲区的位置设置标记（标记当前position的索引位置）</td> 
  </tr> 
  <tr> 
   <td>int position()</td> 
   <td>返回缓冲区的当前位置position</td> 
  </tr> 
  <tr> 
   <td>Buffer position(int n)</td> 
   <td>将设置缓冲区的当前位置为n,并返回修改后的Buffer 对象</td> 
  </tr> 
  <tr> 
   <td>int remaining()</td> 
   <td>返回position 和limit 之间的元素个数</td> 
  </tr> 
  <tr> 
   <td>Buffer reset()</td> 
   <td>将位置position 转到以前设置的mark 所在的位置。</td> 
  </tr> 
  <tr> 
   <td>Buffer rewind()</td> 
   <td>将位置设为为0，取消设置的mark。可重复读。</td> 
  </tr> 
 </tbody> 
</table>

### 2.1.3 缓冲区的数据操作 ###

Buffer 所有子类提供了两个用于数据操作的方法：get() 与put() 方法

获取Buffer 中的数据：

<table> 
 <thead> 
  <tr> 
   <th>byte get()</th> 
   <th>读取单个字节</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>ByteBuffer get(byte[] dst)</td> 
   <td>批量读取多个字节到dst 中</td> 
  </tr> 
  <tr> 
   <td>byte get(int index)</td> 
   <td>读取指定索引位置的字节(不会移动position)</td> 
  </tr> 
 </tbody> 
</table>

放入数据到Buffer 中：

<table> 
 <thead> 
  <tr> 
   <th>ByteBuffer put(byte b)</th> 
   <th>将给定单个字节写入缓冲区的当前位置</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>ByteBuffer put(byte[] src)</td> 
   <td>将src 中的字节写入缓冲区的当前位置</td> 
  </tr> 
  <tr> 
   <td>ByteBuffer put(int index, byte b)</td> 
   <td>将指定字节写入缓冲区的索引位置(不会移动position)</td> 
  </tr> 
 </tbody> 
</table>

### 2.1.4 直接与非直接缓冲区 ###

字节缓冲区要么是直接的，要么是非直接的。如果为直接字节缓冲区，则Java虚拟机会尽最大努力直接在此缓冲区上执行本机I/O操作。也就是说，在每次调用基础操作系统的一个本机I/O操作之前（或之后），虚拟机都会尽量避免将缓冲区的内容复制到中间缓冲区中（或从中间缓冲区中复制内容）。直接缓冲区可以有效减少数据拷贝次数，提升性能，这种方式的学名叫做内存映射。

**传统的io和nio的ByteBuffer\#allocate(capacity)方法获取的缓冲区都是非直接缓冲区，直接字节缓冲区可以通过调用此NIO类的allocateDirect(capacity)方法来创建，此方法返回DirectByteBuffer，此方法返回的直接缓冲区进行分配和取消分配所需成本通常高于非直接缓冲区。**

**直接缓冲区的内容可以驻留在常规的垃圾回收堆之外。因此，它们对应用程序的内存需求量造成的影响可能并不明显，Java的GC也不会对其进行回收**。所以，建议将直接缓冲区主要分配给那些易受基础系统的本机I/O 操作影响的大型、持久的缓冲区。一般情况下，最好仅在直接缓冲区能在程序性能方面带来明显好处时分配它们。

**直接字节缓冲区还可以通过FileChannel\#map()方法将文件区域直接映射到内存中来创建，这就是mmap技术，即内存映射文件技术，该方法返回MappedByteBuffer，通过该方法创建的Buffer可以在垃圾回收时自动回收。DirectByteBuffer实际上就继承自MappedByteBuffer。**

**mmap**是一种内存映射文件的方法，即将一个文件或者其它对象映射到进程的地址空间，实现文件磁盘地址和进程虚拟地址空间中一段虚拟地址的一一对映关系。实现这样的映射关系后，进程就可以采用指针的方式读写操作这一段内存，而系统会自动回写脏页面到对应的文件磁盘上，即完成了对文件的操作而不必再调用read,write等系统调用函数。相反，内核空间对这段区域的修改也直接反映用户空间，从而可以实现不同进程间的文件共享。

**字节缓冲区是直接缓冲区还是非直接缓冲区可通过调用其isDirect()方法来确定。提供此方法是为了能够在性能关键型代码中执行显式缓冲区管理。**

## 2.2 通道（Channel） ##

**通道 Channel 是对原 I/O 包中的流的模拟，可以通过它读取和写入数据。**

通道与流的不同之处在于，流只能在一个方向上移动(一个流必须是 InputStream 或者 OutputStream 的子类)，而通道是双向的，可以用于读、写或者同时用于读写。

**Channel 本身不能直接访问数据，Channel 只能与Buffer 进行交互。**

### 2.2.1 Channel的实现 ###

Java 为Channel 接口提供的最主要实现类如下：

1.  FileChannel：用于读取、写入、映射和操作文件的通道。
2.  DatagramChannel：通过UDP 读写网络中的数据通道。
3.  SocketChannel：通过TCP 读写网络中的数据，客户端通道。
4.  ServerSocketChannel：可以监听新进来的TCP 连接，对每一个新进来的连接都会创建一个SocketChannel，服务端通道。

#### 2.2.2 获取通道 ####

1.  获取通道的一种方式是对支持通道的对象调用getChannel()方法。支持通道的类如下：
2.  本地IO的相关类，FileInputStream、FileOutputStream、RandomAccessFile，获取FileChannel。
3.  网络IO相关类，DatagramSocket获取DatagramChannel，Socket获取SocketChannel，ServerSocket获取ServerSocketChannel
4.  在JDK1.7中的NIO.2中，新增了Files\#newByteChannel()获取FileChannel的方式，以及通过各个通道实现类的静态方法open()打开并返回指定通道的方式。

# 3 选择器Selector #

NIO 常常被叫做非阻塞 IO，主要是因为 NIO 在网络通信中的非阻塞特性被广泛使用。

Java NIO 实现了 IO 多路复用中的 Reactor 模型，一个线程使用一个选择器 Selector 通过轮询的方式去监听多个通道 Channel 上的事件，从而让一个线程就可以处理多个事件。

通过配置监听的通道 Channel 为非阻塞，那么当 Channel 上的 IO 事件还未到达时，就不会进入阻塞状态一直等待，而是继续轮询其它Channel，找到IO事件已经到达的 Channel 执行。

因为创建和切换线程的开销很大，因此使用一个线程来处理多个事件而不是一个线程处理一个事件，对于 IO 密集型的应用具有很好地性能。

应该注意的是，只有套接字 Channel 才能配置为非阻塞，而 FileChannel 不能，为 FileChannel 配置非阻塞也没有意义。  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5YiYSmF2YQ_size_20_color_FFFFFF_t_70_g_se_x_16_pic_center]  
选择器（Selector）是SelectableChannel 对象的多路复用器，Selector可以同时监控多个SelectableChannel 的IO 状况，也就是说，利用Selector 可使一个单独的线程管理多个Channel。Selector 是非阻塞IO 的核心。

SelectableChannel 的结构如下图：  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5YiYSmF2YQ_size_20_color_FFFFFF_t_70_g_se_x_16_pic_center 1]

## 3.1 创建Selector ##

通过调用Selector\#open() 方法创建一个Selector。

Selector st = Selector.open();

## 3.2 向选择器注册通道 ##

SelectionKey sk = SelectableChannel.register(Selector sel, int ops)

该方法的返回值是一个SelectionKey对象，SelectionKey表示一个Selector和Channel 之间的注册关系。当Channel 注册到Selector 上时，就相当于确立了两者的服务关系，而SelectionKey 就是这个契约。当Selector或者Channel被关闭时， 它们对应的SelectionKey 就会失效。

当调用register方法将通道注册选择器时，选择器对通道的监听事件需要通过第二个参数ops 指定。可以监听的事件类型（可使用SelectionKey 的四个常量表示）：

1.  读: SelectionKey.OP\_READ （1）
2.  写: SelectionKey.OP\_WRITE （4）
3.  连接: SelectionKey.OP\_CONNECT（8）
4.  接收: SelectionKey.OP\_ACCEPT （16）

若注册时不止监听一个事件，则可以使用“位或”操作符连接。例：

int ops= SelectionKey.OP_ACCEPT| SelectionKey.OP_CONNECT;

SelectionKey的方法：

<table> 
 <thead> 
  <tr> 
   <th>方法</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>int interestOps()</td> 
   <td>获取感兴趣事件集合</td> 
  </tr> 
  <tr> 
   <td>int readyOps()</td> 
   <td>获取通道已经准备就绪的操作的集合</td> 
  </tr> 
  <tr> 
   <td>SelectableChannel channel()</td> 
   <td>获取注册通道</td> 
  </tr> 
  <tr> 
   <td>Selector selector()</td> 
   <td>返回选择器</td> 
  </tr> 
  <tr> 
   <td>boolean isReadable()</td> 
   <td>检测Channal 中读事件是否就绪</td> 
  </tr> 
  <tr> 
   <td>boolean isWritable()</td> 
   <td>检测Channal 中写事件是否就绪</td> 
  </tr> 
  <tr> 
   <td>booleanisConnectable()</td> 
   <td>检测Channel 中连接是否就绪</td> 
  </tr> 
  <tr> 
   <td>booleanisAcceptable()</td> 
   <td>检测Channel 中接收是否就绪</td> 
  </tr> 
 </tbody> 
</table>

## 3.3 监听事件 ##

int num = selector.select();

使用 select()方法来监听到达的事件，它会一直阻塞直到有至少一个事件到达。这个方法应该被循环调用。

## 3.4 Selector 的常用方法 ##

<table> 
 <thead> 
  <tr> 
   <th>方法</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Set keys()</td> 
   <td>所有的SelectionKey 集合。代表注册在该Selector上的Channel</td> 
  </tr> 
  <tr> 
   <td>selectedKeys()</td> 
   <td>被选择的SelectionKey 集合。返回此Selector的已选择键集</td> 
  </tr> 
  <tr> 
   <td>intselect()</td> 
   <td>监控所有注册的Channel，当它们中间有需要处理的IO 操作时，该方法返回，并将对应得的SelectionKey 加入被选择的SelectionKey 集合中，该方法返回这些Channel 的数量。</td> 
  </tr> 
  <tr> 
   <td>int select(long timeout)</td> 
   <td>可以设置超时时长的select() 操作</td> 
  </tr> 
  <tr> 
   <td>intselectNow()</td> 
   <td>执行一个立即返回的select() 操作，该方法不会阻塞线程</td> 
  </tr> 
  <tr> 
   <td>Selectorwakeup()</td> 
   <td>使一个还未返回的select() 方法立即返回</td> 
  </tr> 
  <tr> 
   <td>void close()</td> 
   <td>关闭该选择器</td> 
  </tr> 
 </tbody> 
</table>

# 4 NIO非阻塞网络通信案例 #

## 4.1 非阻塞式TCP通信 ##

Java NIO中的SocketChannel是一个连接到TCP网络套接字的通道。Java NIO中的ServerSocketChannel是一个可以监听新进来的TCP连接的通道，就像标准IO中的ServerSocket一样。

TcpClient：

/**
     * @author lx
     */
    public class TcpClient {
        
        public static void main(String[] args) throws IOException {
        
            //1. 获取通道
            SocketChannel sChannel = SocketChannel.open(new InetSocketAddress("127.0.0.1", 9898));
            //2. 切换非阻塞模式
            sChannel.configureBlocking(false);
            //3. 分配指定大小的缓冲区
            ByteBuffer buf = ByteBuffer.allocate(1024);
            //4. 发送数据给服务端
            Scanner scan = new Scanner(System.in);
            while (scan.hasNext()) {
        
                String str = scan.next();
                buf.put((new Date().toString() + "\n" + str).getBytes());
                buf.flip();
                sChannel.write(buf);
                buf.clear();
            }
            //5. 关闭通道
            sChannel.close();
        }
    }

TcpServer：

/**
     * @author lx
     */
    public class TcpServer {
        
        public static void main(String[] args) throws IOException {
        
            //1. 获取通道
            ServerSocketChannel ssChannel = ServerSocketChannel.open();
            //2. 切换非阻塞模式
            ssChannel.configureBlocking(false);
            //3. 绑定连接
            ssChannel.bind(new InetSocketAddress(9898));
            //4. 获取选择器
            Selector selector = Selector.open();
            //5. 将通道注册到选择器上, 并且指定“监听接收事件”
            ssChannel.register(selector, SelectionKey.OP_ACCEPT);
            //6. 轮询式的获取选择器上的一个已经“准备就绪”的事件
            while (selector.select() > 0) {
        
                //7. 获取当前选择器中所有注册的“选择键(已就绪的监听事件)”
                Iterator<SelectionKey> it = selector.selectedKeys().iterator();
                while (it.hasNext()) {
        
                    //8. 获取准备就绪的事件
                    SelectionKey sk = it.next();
                    //9. 判断具体是什么事件准备就绪
                    if (sk.isAcceptable()) {
        
                        //10. 若“接收就绪”，获取客户端连接
                        SocketChannel sChannel = ssChannel.accept();
                        //11. 切换非阻塞模式
                        sChannel.configureBlocking(false);
                        //12. 将该通道注册到选择器上
                        sChannel.register(selector, SelectionKey.OP_READ);
                    } else if (sk.isReadable()) {
        
                        //13. 获取当前选择器上“读就绪”状态的通道
                        SocketChannel sChannel = (SocketChannel) sk.channel();
                        //14. 读取数据
                        ByteBuffer buf = ByteBuffer.allocate(1024);
                        int len = 0;
                        while ((len = sChannel.read(buf)) > 0) {
        
                            buf.flip();
                            System.out.println(new String(buf.array(), 0, len));
                            buf.clear();
                        }
                    }
                    //15. 取消选择键 SelectionKey
                    //这个非常重要，当处理完一个SelectionKey后，务必将其从集合内删除
                    //否则下次就会重复处理相同的SelectionKey
                    it.remove();
                }
            }
        }
    }

## 4.2 非阻塞式UDP通信 ##

Java NIO中的DatagramChannel是一个能收发UDP包的通道。

UdpClient：

/**
     * @author lx
     */
    public class UdpClient {
        
        public static void main(String[] args) throws IOException {
        
            DatagramChannel dc = DatagramChannel.open();
            dc.configureBlocking(false);
            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
            Scanner scanner = new Scanner(System.in);
            while (scanner.hasNext()) {
        
                String next = scanner.next();
                byteBuffer.put((new Date().toString() + "\n" + next).getBytes());
                byteBuffer.flip();
                dc.send(byteBuffer, new InetSocketAddress("127.0.0.1", 9898));
                byteBuffer.clear();
            }
            dc.close();
        }
    }

UdpServer：

/**
     * @author lx
     */
    public class UdpServer {
        
        public static void main(String[] args) throws IOException {
        
            DatagramChannel dc = DatagramChannel.open();
            dc.configureBlocking(false);
            dc.bind(new InetSocketAddress(9898));
            Selector selector = Selector.open();
            dc.register(selector, SelectionKey.OP_READ);
            while (selector.select() > 0) {
        
                Iterator<SelectionKey> selectionKeyIterator = selector.selectedKeys().iterator();
                while (selectionKeyIterator.hasNext()) {
        
                    SelectionKey next = selectionKeyIterator.next();
                    if (next.isReadable()) {
        
                        ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                        dc.receive(byteBuffer);
                        byteBuffer.flip();
                        System.out.println(new String(byteBuffer.array(), 0, byteBuffer.limit()));
                        byteBuffer.clear();
                    }
                }
                selectionKeyIterator.remove();
            }
        }
    }

> 如有需要交流，或者文章有误，请直接留言。另外希望点赞、收藏、关注，我将不间断更新各种Java学习博客！

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5YiYSmF2YQ_size_20_color_FFFFFF_t_70_g_se_x_16_pic_center]: /images/20220828/49c2b909f9de4c16bc5e6f4fdce16a61.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5YiYSmF2YQ_size_20_color_FFFFFF_t_70_g_se_x_16_pic_center 1]: /images/20220828/a7efd868076a463a860d726756c1164a.png