NIO 入门

NIO 入门

目录

• NIO 入门
  • 输入/输出：概念性描述
    • 流与块的比较
  • 通道和缓冲区
    • 缓冲区
    • 缓冲区类型
    • 通道
    • 通道类型
  • 实践：NIO的读与写
    • 从文件中读取
    • 写入文件
    • 读写结合
  • 缓冲区内部细节
    • 状态变量
    • 访问方法
      • get()方法
      • put()方法
      • 类型化的get()和put()方法
  • 关于缓冲区的额外内容
    • 缓冲区分配和包装
    • 缓冲区分片
    • 只读缓冲区
    • 直接和间接缓冲区
    • 内存映射文件I/O
  • 分散和聚集
    • 分散/聚集I/O
    • 应用
  • 文件锁定
    • 锁定文件
    • 可移植性
  • 连网和异步I/O
    • 异步I/O
    • 示例
  • 字符集
    • 编码
• NIO浅析
  • 传统BIO模型(Blocking I/O)
  • NIO工作原理
    • 常见I/O模型

输入/输出：概念性描述

传统IO：

• 使用流的方式完成IO。
• 所有I/O被视为单个的字节来移动。
• 通过Stream的对象一次移动一个字节。

流与块的比较

• 传统IO与NIO的区别在于数据的打包和传输的方式。
• 传统IO ==> 以流的方式处理数据。
• NIO ==> 以块的方式处理数据。
• 流式I/O系统：
  • 一次一个字节地处理数据。
  • 一个输入（输出）流产生（消费）一个字节的数据。
• 块式I/O系统：
  • 每个操作都是在一步中产生或消费一个数据块。
  • 比流式字节处理速度快。

---

通道和缓冲区

• 通道：到任何目的地（从任何目的地来）的所有数据都必须通过Channel对象。
• 缓冲区：一个Buffer是一个容器对象。发送给一个通道的所有对象都放在缓冲区中（从通道读取数据也一样）。

缓冲区

• Buffer是一个对象，包含要读出或写入的数据。
• NIO中的所有数据都用缓冲区进行处理。读入的数据放在缓冲区，写出的数据写到缓冲区中。
• 缓冲区本质是一个字节数组（或其他类型数组）。

缓冲区类型

• 常用的缓冲区类型为ByteBuffer。
• 每种基本Java类型都有一个缓冲区类型：
  • ByteBuffer
  • CharBuffer
  • ShortBuffer
  • IntBuffer
  • LongBuffer
  • FloatBuffer
  • DoubleBuffer

通道

• Channel是一个对象，用来读取和写入数据。
• 将数据写入到缓冲区而非通道中。
• 将数据从通道写入缓冲区，再由缓冲区获取数据。

通道类型

• 流：单向的，只在一个方向流动。(InputStream，OutputStream)
• 通道：双向的，可用于读，写或同时读写。

---

实践：NIO的读与写

• 读取：
  • 创建缓冲区。
  • 通过通道将数据读到缓冲区中。
• 写入：
  • 创建缓冲区。
  • 将数据写入缓冲区。
  • 让通道使用该数据执行写入操作。

从文件中读取

步骤：

1. 从FileInputStream获取Channel。
2. 创建Buffer。
3. 使用Channel将数据读到Buffer中。

示例：

// 获取通道
FileInputStream fin = new FileInputStream("test.md");
FileChannel fc = fin.getChannel();

// 创建缓冲区
ByteBuffer buffer = ByteBuffer.allocate(1024);

// 利用通道将数据读到缓冲区
fc.read(buffer);

写入文件

步骤：

1. 从FileOutputStream获取Channel。
2. 创建Buffer。
3. 将数据放入读入Buffer中。
4. 设置指针为缓冲区开始位置。
5. 通过Channel将数据从缓冲区写出。

示例：

// 创建输出流对象
FileOutputStream fout = new FileOutputStream();

// 由输出流对象获得通道
FileChannel fc = fout.getChannel();

// 创建缓冲区
ByteBuffer buffer = ByteBuffer.allocate(1024);

// 将数据放到缓冲区中
for(int i = 0; i < message.length;++i){
    buffer.put(message[i]);
}

// 设置指针指向缓冲区开始
buffer.flip();

// 利用通道将缓冲区数据写出
fc.write(buffer);

读写结合

将一个文件的所有内容拷贝到另外一个文件中。

步骤：

1. 创建缓冲区。
2. 获取输入，输出通道。
3. 将数据通过输入通道读到缓冲区。
4. 利用输出通道将缓冲区数据写出到目标文件。

示例：

FileInputStream fin = new FileInputStream(infile); // 输入流对象
FileOutputStream fou = new FileOutputStream(outfile); // 输出流对象

FileChannel fcin = fin.getChannel(); // 获取输入通道
FileChannel fcout = fou.getChannel(); // 获取输出通道

ByteBuffer buffer = ByteBuffer.allocate(1024); // 创建缓冲区

// 拷贝文件
while (true){
    buffer.clear(); // 清空缓冲区

    int r = fcin.read(buffer); // 通过输入通道将数据读入

    if (r == -1) // 若到文件末尾，则退出
        break;

    buffer.flip(); // 将指针至到缓冲区开始，从开始位置输出数据

    fcout.write(buffer); // 通过输出通道将数据从缓冲区写出到文件
}

---

缓冲区内部细节

缓冲区有两个重要的组件:

• 状态变量
• 访问方法(accessor)

状态变量

缓冲区有三个状态变量指示当前的状态:

• position
• limit
• capacity

Position

• 缓冲区实际上是一个数组.
• position用来指示下一次操作所指向的数组索引位置.
• 若为读取数据,则position指示下一个读入的数据应该放在数组的位置.
• 若为写出数据,则position指示下一个写出的数据在数组的位置.

Limit

• 若为读取数据,limit指示position读入数据的位置不能超过限制.
• 若为写出数据,limit指示position写出的数据不能超过的位置.
• position不能超过limit指示的位置.

Capacity

• 存储在缓冲区的最大数据容量,即底层数组的大小.
• limit不能超过capacity.

示例

1. 创建一个大小为n的缓冲区.则此时capacity为n,limit为n,position为0.
2. 第一个读入a个字节后,position指向位置a,其他保持不变.
3. 第二次读入b个字节后,position指向位置a+b,其他保持不变.
4. 要将缓冲区的数据输出,先调用flip().其将limit设置为position指向的位置a+b,position设置为0.
5. 第一次写出a个字节,则此时position指向位置a.
6. 第二次写出b个字节,则此时position指向位置a+b.
7. 调用clear()方法,将清空缓冲区,并将position设置为0,limit设置为capacity.

访问方法

get()方法

分类:

1. byte get() ==> 获取单个字符,操作影响positon
2. ByteBuffer get(byte dst[]); ==> 将一组字符读入数组中,操作影响position
3. ByteBuffer get(byte dst[], int offset, int length); ==> 将一组字符读入数组中,操作影响position
4. byte get(int index); ==> 从特定位置获取字符,与position无关

put()方法

分类:

1. ByteBuffer put(byte b); ==> 写入单个字节,影响position
2. ByteBuffer put(byte src[]); ==> 写入一组字节,影响position
3. ByteBuffer put(byte src[], int offset, int length); ==> 写入一组字节,影响position
4. ByteBuffer put(ByteBuffer src); ==> 从ByteBuffer写入到当前ByteBuffer,影响position
5. ByteBuffer put(int index, byte b); ==> 将字节写入到指定的位置,与position无关

类型化的get()和put()方法

对于不同的类型有不同的方法.

---

关于缓冲区的额外内容

缓冲区分配和包装

• 在创建缓冲区时,可使用静态方法allocate(缓冲区大小)分配缓冲区.
• 也可以使用wrap()方法将已有的数组转换成缓冲区.
• 若使用wrap()方法获得缓冲区,则通过原数组也可访问底层数据.

缓冲区分片

• slice()由已有的缓冲区创建一个子缓冲区.
• 子缓冲区与原缓冲区的部分共享数据.
• 通过对子缓冲区操作,将影响原缓冲区中的数据.

只读缓冲区

• 只读缓冲区:可以读取,不能向其写入.
• 使用asReadOnlyBuffer()方法将普通缓冲区转换为只读缓冲区.
• 方法返回的缓冲区与原缓冲区完全相同,但是只读.
• 返回的缓冲区与原缓冲区共享数据,原缓冲区的修改导致只读缓冲区受到影响.
• 不能将只读缓冲区转换为可写缓冲区.

直接和间接缓冲区

直接缓冲区:
加快I/O速度,以特殊的方式分配其内存的缓冲区.

给定一个直接字节缓冲区，Java 虚拟机将尽最大努力直接对它执行本机 I/O 操作。也就是说，它会在每一次调用底层操作系统的本机 I/O 操作之前(或之后)，尝试避免将缓冲区的内容拷贝到一个中间缓冲区中(或者从一个中间缓冲区中拷贝数据)。

内存映射文件I/O

通过使文件中的数据以内存数组的内容来完成.
一般只有实际读取或写入的部分才会送入内存中.
其提供底层操作系统的机制调用.

示例

MappedByteBuffer mbb = fc.map(FileChannel.MapMode.READ_WRITE,0,1024);
// 将FileChannel的前1024个字节映射到内存中
// 返回值为MappedByteBuffer,为ByteBuffer子类

---

分散和聚集

• 分散/聚集I/O使用多个缓冲区保存数据.
• 分散读取:将数据读到一个缓冲区数组.
• 聚集写入:向缓冲区数组写入数据.

分散/聚集I/O

两个接口:

• ScatteringByteChannel
• GatheringByteChannel

ScatteringByteChannel的两个读方法

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

特点:

• 分散读取:依次填充每个缓冲区.填满一个缓冲区后,填充下一个缓冲区.

聚集写入的两个写方法

• long write(ByteBuffer[] srcs);
• long write(ByteBuffer[] srcs, int offset, int length);

应用

有一个网络应用,每个消息被划分成固定长度的头部和固定长度的正文.

创建一个容纳头部的缓冲区和一个容纳正文的缓冲区.

---

文件锁定

文件锁定:不阻止任何形式的数据访问,而是通过锁的共享和获得运行不同的部分相互协调.

共享锁:其他人可以获得共享锁,但不能获得排它锁.
排它锁:其他人不能获得同一文件的锁.

锁定文件

• 使用写方式打开文件.
• 获得对应文件的锁.

示例:

RandomAccessFile raf = new RandomAccessFile("test.md","rw");
FileChannel fc = raf.getChannel();
FileLock lock = fc.lock(start,end,false);

可移植性

原则:

• 只使用排它锁.
• 将所有的锁视为劝告式的(advisory).

---

连网和异步I/O

异步I/O

• 没有阻塞地读写数据.
• 通过注册特定I/O事件(可读数据的到达,新套接字连接).当发生注册事件,系统发出通知.
• 异步I/O可以监听任意数量的通道上的事件而不用额外的线程.

示例

private int ports[];
private ByteBuffer echoBuffer = ByteBuffer.allocate( 1024 );

private void go() throws IOException {
    // 创建一个selector
    // 是注册对I/O事件感兴趣的地方,当事件发送时,selector发出通知
    Selector selector = Selector.open();
    
    // 对每个端口打开一个监听器,并注册到selector中
    for (int i=0; i<ports.length; ++i) {
      // 为监听每个端口,每个端口需要一个ServerSocketChannel
      ServerSocketChannel ssc = ServerSocketChannel.open();
      // 设置为非阻塞式
      ssc.configureBlocking( false );
      // 新建一个socket
      ServerSocket ss = ssc.socket();
      // 新建socket地址
      InetSocketAddress address = new InetSocketAddress( ports[i] );
      // socket绑定端口地址
      ss.bind( address );
      // 将ServerSocketChannels注册到selector上
      // 第一个参数是selector,第二个参数是指定监听的事件
      // 返回值表示通道在此selector上的注册,当通知发生事件时,是提供该事件的selectionKey进行的
      SelectionKey key = ssc.register( selector, SelectionKey.OP_ACCEPT );
    
      System.out.println( "Going to listen on "+ports[i] );
    }
    
    while (true) {
      // 阻塞,直到一个或多个事件发生
      // 返回发生的事件数
      int num = selector.select();
      // 返回所有事件对应的selectedKey的集合
      Set selectedKeys = selector.selectedKeys();
      Iterator it = selectedKeys.iterator();
      //对于每个事件的处理
      while (it.hasNext()) {
        SelectionKey key = (SelectionKey)it.next();
        // 获取selectKey对应事件的类型,若有新连接,则接收
        if ((key.readyOps() & SelectionKey.OP_ACCEPT)
          == SelectionKey.OP_ACCEPT) {
          // 创建ServerSocketChannel
          ServerSocketChannel ssc = (ServerSocketChannel)key.channel();
          SocketChannel sc = ssc.accept(); // 接受连接
          sc.configureBlocking( false ); // 设置为非阻塞式
    
          // 接收完后,更新新的selectionKey,用来接收新连接
          SelectionKey newKey = sc.register( selector, SelectionKey.OP_READ ); // 注册为读取
          it.remove(); // 将处理完的selectionKey从集合中删除,否则会被再次处理
    
          System.out.println( "Got connection from "+sc );
        }
        // 若套接字的数据到达,则接收数据
        else if ((key.readyOps() & SelectionKey.OP_READ)
          == SelectionKey.OP_READ) {
          // 获取处理的通道
          SocketChannel sc = (SocketChannel)key.channel();
    
          int bytesEchoed = 0;
          while (true) {
            echoBuffer.clear();
            // 获取读取结果
            int r = sc.read( echoBuffer );
            // 小于等于0则传输结束
            if (r<=0) {
              break;
            }
    
            echoBuffer.flip();
            // 写出缓冲区数据
            sc.write( echoBuffer );
            bytesEchoed += r;
          }
    
          System.out.println( "Echoed "+bytesEchoed+" from "+sc );
          // 移除selectionKey,避免重复处理
          it.remove();
        }

---

字符集

Charset:十六位Unicode字符序列与字节序列之间的一个命名映射.

编码

读文本:CharsetDecoder.(逐位将字符转换为char值)
写文本:CharsetEncoder.(将字符转换为位)

Java支持的字符编码:

• US-ASCII
• ISO-8859-1
• UTF-8
• UTF-16BE
• UTF-16LE
• UTF-16

示例:

// 创建字符集实例
Charset latin1 = Charset.forName( "ISO-8859-1" );
CharsetDecoder decoder = latin1.newDecoder(); // 字符集对应的解码器
CharsetEncoder encoder = latin1.newEncoder(); // 字符集对应的编码器

CharBuffer cb = decoder.decode( inputData ); // 将字符数据解码,生成缓冲区

ByteBuffer outputData = encoder.encode( cb ); // 将缓冲区数据编码

outc.write( outputData ); // 输出编码后的数据

inf.close();
outf.close();

---

NIO浅析

NIO(Non-blocking I/O):同步非阻塞的I/O模型,I/O多路复用的基础.

传统BIO模型(Blocking I/O)

传统服务器端同步阻塞I/O处理:

ExecutorService executor = Excutors.newFixedThreadPollExecutor(100); // 线程池
ServerSocket serverSocket = new ServerSocket(); // 创建新socket
serverSocket.bind(8088); // socket绑定端口
while(!Thread.currentThread.isInturrupted){ // 线程循环等待新连接
  Socket socket = serverSocket.accept(); // 接收新连接
  executor.submit(new ConnectIOnHandler(socket)); // 为新连接创建一个新线程
}

class ConnectIOnHandler extends Thread{
  private Socket socket;
  public ConnectIOnHandler(Socket socket){
    this.socket = socket;
  }
  public void run(){
    // 循环处理读写事件
    while(!Thread.currentThread.isTnturrupted()&&!socket.isClosed()){
      String someThing = socket.read(); // 读取数据
      if(someThing != null){
        // 处理数据
        socket.write(); // 写数据
      }
    }
  }
}

特点:

• 每个连接对应一个线程.
• 多个线程是因为socket的accpet(),read()和write()是**同步阻塞的,即:每个连接处理I/O时是阻塞的.
• 模型简单,适用于连接数较少的情况.
• 线程的创建和销毁成本高.
• 线程占用内存大.
• 线程间的切换成本高(保留上下文,系统调用,切换时间可能大于执行时间==>load高,sy使用高,系统不可用).
• 造成锯齿状系统负载(大量阻塞线程使系统负载压力大).

---

NIO工作原理

常见I/O模型

I/O的两个阶段:

• 等待就绪
• 操作

• BIO:read()方法若没有收到数据,则一直阻塞,直到收到数据后返回数据.
• NIO:若有数据,则将数据读到内存,并返回;否则直接返回0,不阻塞.
• AIO:等待就绪是非阻塞的,读取数据也是异步的.

参考:

• NIO 入门
• Java NIO浅析