NIO

本篇文章主要来自:NIO-IBM,文中的代码可以从这里下载.

NIO(new input output)是从java1.4引入的,NIO提供了基于块的(block-oriented)的,.和原来的IO模型(java.io.*)比起来速度上更快.NIO将比较耗时的IO操作交给了操作系统,如填充缓冲区等工作交还给了操作系统,因此速度更快.(这里的话,我认为的意思可能就是在java层面的缓冲区填充效率较低).

Streams vs blocks

NIO和老的IO之间的区别就是数据是如何传输的.在原来的IO模型中,操作的对象都是流式(stream)的,然而NIO操作的是块的.基于stream的io系统来说,每次处理的都是一个字节,比如说inputstream读取一个字节,outputstream消耗一个字节.所以呢比较慢.

NIO来说,处理的数据都是以块为基本单位的.读取一个块或者消耗一个块,所以呢NIO比stream io更快.不过在jdk1.4中stream io和nio被很好的集成在一块了.比如说java.io.*中也包含着基于块的数据读取.

Buffer and channel

channel和buffer是nio中关键的东西.channels就和stream io中的stream类似,对于数据的操作我们都是使用channel的.如在stream io中,我们从文件读取数据用的是FileInputStream,在NIO中使用的是就是FileChannel.

从channel中读取的数据都要放到buffer中,与之类似的,往channel中写数据,也需要先放到buffer中.最常用的buffer类型是bytebuffer,在NIO中还有其他几类buffer:

• ByteBuffer
• CharBuffer
• ShortBuffer
• IntBuffer
• LongBuffer
• FloatBuffer
• DoubleBuffer

channel与stream的api不同的是,stream api是单向的,也就是说,我们读取数据的时候用的是inputstream,写数据的时候用的是outputstream.但是channel类可以写入,也可以读取,是双向的,这和unix中的pipe是一个意思.

Reading from a file

对于IO操作来说,最常用的功能就是从文件中读取数据,基本的来说,NIO的io操作都分为下面几个步骤:

1. 新建stream对象
2. 从该对象中获得channel对象
3. 然后新建一个buffer
4. 从channel中读取数据放到buffer中.

分配了一个1024字节的缓冲区,将数据从channel中读取到buffer中.代码如下:

public class UserBuffer {
    public static void main(String[] args) throws IOException {
        String dir = "/home/ygj/Desktop/hello.s";
        FileInputStream file = new FileInputStream(dir);
        FileChannel channel = file.getChannel();
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        channel.read(buffer);
        buffer.flip();
        while(buffer.hasRemaining()) {
            byte b = buffer.get();
            System.out.printf("%c",b);
        }
        file.close();
    }
}

对于代码中的flip()函数暂且不要理会,后面会解释.

Writing to file

写的过程与之类似，只不过我们要从FileOutputStream中获得channel，在这了我们将hello写入到hello.txt中,代码如下:

public class UserBuffer {
    public static void main(String[] args) throws IOException {
        String dir = "/home/ygj/Desktop/hello.txt";
        byte[] data = new byte[]{'h','e','l','l','o'};
        FileOutputStream outputStream = new FileOutputStream(dir);
        FileChannel fc = outputStream.getChannel();
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        for (int i = 0; i < data.length; i++) {
            buffer.put(data[i]);
        }
        buffer.flip();
        fc.write(buffer);
        fc.close();
    }
}

Copying file

目前来说,复制文件看起来还是不容易的,我们只涉及了缓冲区的大小,那么如果文件太大了缓冲区放不下操作呢?

接下来就讲一下如何以NIO来操作复制文件.

其他的代码不表,只展示一些关键的代码:

while (true) {
    buffer.clear(); //清空缓冲区
    int r = ch1.read(buffer);
    if (r == -1) //判断是否还有数据要读取
        break;
    buffer.flip();
    ch2.write(buffer); //写入到文件中国
}

在每次读取之后,都清空缓冲区.

Buffer Internals

buffer就是一个数组,如果有C语言的变成经验,IO操作往往需要传入到缓冲区中,缓冲区就是一个数组.我们需要一些变量来管理缓冲区,比如说,下一个字节应该写到哪里,缓冲区的容量是多少等等.

NIO的buffer主要有三个变量:

1. position
2. limit
3. capcity

position表示着已经有多少数据写入到了buffer中,更准确的来说,position指定了下一个要写入的下标.比如说,现在写入了3个字节,那么position就是3.因为数组中的[0,1,2]已经被写入了.读也是差不多的,意思.比如说读了三个字节的数据,那么此时position就是3.

limit 当前操作最多可以达到缓冲区的哪里,这个可能稍微有点拗口,接下来的例子就明白了

capcity表明了buffer总共大小是多少,就是数组的大小

下面举一个例子:

    [0,1,2,3,4,5,6,7]
     ^             ^
 position         limit
                  capcity

8个元素的缓冲区,往里面写入三个字节的数据,那么下一次position就会指向3:

[0,1,2,3,4,5,6,7]
       ^
      position

然后,如果我们要从缓冲区(此时有三个字节的数据)中读取数据,调用flip()函数,它做以下的事情:

1. 将limit修改为此时的position,来限定我们能读取多少数据.
2. position修改为了0.

就变成如下:

        [0,1,2,3,4,5,6,7]
         ^     ^
     position  limit

这样一来,limit的作用就很好懂了,如果此时我们要开始读取数据,limit就可以很好地限制了接下来数据读取的范围.

最后clear()函数的作用是将limit和position归为.

    [0,1,2,3,4,5,6,7]
     ^             ^
    position     limit

Access Method

既然buffer是一个数组,那么我们很自然的可以根据下标来访问或者读取缓冲区的内容.对于get()来说,有几个方法是相对的(relative),还有一个是绝对的(absolute),相对和觉得的意思是: get()是否依赖于position和limit.

byte get(); //从当前postion读取一个字节
ByteBuffer get(byte dst[]);//读取一堆buffer的当前position数据到dst中
ByteBuffer get(byte dst[],int offset, int length); //从buffer中的position处读取length字节的数据到dst的offset起始的地方.
byte get(int index); //从buffer的index处读取一个字节的内容

与之相类似的,put()也有这样的形式:

ByteBuffer put( byte b );
ByteBuffer put( byte src[] );
ByteBuffer put( byte src[], int offset, int length );
ByteBuffer put( ByteBuffer src );
ByteBuffer put( int index, byte b ); //绝对的

此外还有一些方法来获得buffer中的数据:

byte getChar(); //相对
byte getChar(int index); //绝对

此外还有getInt,getLong等等.最后,介绍过buffer内部的几个遍历之后,再来看一下之前复制文件数据的操作:

while (true) {
    buffer.clear(); //limit和position归位
    int r = ch1.read(buffer); //read会修改position
    if (r == -1)
        break;
    buffer.flip(); //limit = position,将position = 0
    ch2.write(buffer); //所以可以正常的写入数据
}

Buffer allocation and wrapping

创建一个buffer的方式,可以使使用ByteBuffer.allocate(int size)这样的形式.

ByteBuffer buffer = ByteBuffer.allocate(1024);

或者是自己创建一个数组:

byte array[] = new byte[1024];
ByteBuffer buffer = ByteBuffer.wrap(array);

注意,如果使用这种方式的话,对于array的访问可以是通过buffer的函数,或者是直接使用下标.

Buffer slicing and data sharing

slice()的作用是从现有的buffer中切割一部分出来.创建一个新的子缓存(sub-buffer).子缓存和原来的缓存使用的是相同的底层数组,也就是说对于子缓存的修改也会影响到原来的缓存.

下面是buffer slice的操作:

buffer.position(3);
buffer.limit(7);
ByteBuffer slice = buffer.slice();

用下面代码来说明这个问题:

ByteBuffer buffer = ByteBuffer.allocate(10);
for (int i = 0; i < buffer.capacity(); i++) { //初始化缓冲区
    buffer.put((byte) i);
}
//以postion = 3,limit = 7创建一个sub-buffer
buffer.position(3);
buffer.limit(7);
ByteBuffer slice = buffer.slice();
for (int i = 0; i < slice.capacity(); i++) { //修改sub buffer的内容
    byte b = slice.get();
    b *= 10;
    slice.put(i,b);
}
buffer.position(0);
buffer.limit(buffer.capacity());
for (int i = 0; i < buffer.capacity(); i++) { //输出buffer中的内容
    System.out.println(buffer.get());
}
// output: 0 1 2 30 40 50 60 7 8 9

Readonly-buffer

只读的缓存使得只能从buffer中读取数据,但是不能写入.代码如下:

public class UserBuffer {
    public static void main(String[] args) throws IOException {
        ByteBuffer buffer = ByteBuffer.allocate(10);
        for (int i = 0; i < buffer.capacity(); i++) {
            buffer.put((byte) i);
        }
        buffer.flip(); //写入之后,如果需要读取数据,flip一下
        buffer.asReadOnlyBuffer(); //转为知道只读的buffer
        while(buffer.hasRemaining()) {
            System.out.println(buffer.get());
        }
    }
}

Direct Buffer

可以通过allocateDirect()函数来申请一个direct buffer.对于direct buffer来说,在操作系统的IO操作之前可以避免将缓冲区的数据复制到一个临时的buffer.简单来说,就是少了缓冲区复制的操作.但是它并不是位于堆内的内存区域,所以GC管不到它,对于direct buffer来说有更高的分配以及回收的开销.

在io操作中,操作系统从外部设备(网卡,硬盘等)的数据往往是先放在内核的缓冲区中,然后在复制到用户空间中的缓冲区.这就是上面提到的缓冲区复制的操作.

文档中提到:

A direct byte buffer may be created by invoking the allocateDirect factory method of this class. The buffers returned by this method typically have somewhat higher allocation and deallocation costs than non-direct buffers. The contents of direct buffers may reside outside of the normal garbage-collected heap, and so their impact upon the memory footprint of an application might not be obvious.

Memory-mapped file IO

这个就是posix中的mmap().Memory-mapped file IO能够将文件或者外设(devices)映射到内存中,但是它并不是将整个文件都映射到内存中的,而是按需的(on-demond).

mmio操作会直接修改硬盘中的文件内容.下面是一个mmio的例子:

public class UserBuffer {
    public static void main(String[] args) throws IOException {
        String dir = "/home/ygj/Desktop/hello.c";
        RandomAccessFile file = new RandomAccessFile(dir,"rw");
        FileChannel channel = file.getChannel();
        MappedByteBuffer mmap = channel.map(FileChannel.MapMode.READ_WRITE,0,1024);
        mmap.put(0, (byte) 'h');
        mmap.put(2,(byte)'a');
        file.close();
    }
}
//会修改hello.c中的第一个和第三个字节的内容.

IO multiplexing

虽然原文没有明确的说本节的内容是io多路复用的东西.不过,从操作系统的层面来说,它确实就是.

同步IO网络模型

在经典的网络模型中,创建ServerSocket对象之后,调用它的accept()函数来等待来自客户端的链接，当一个新的连接建立的时候，再开一个新的线程来处理该请求，大致代码如下:

Socket socket = serverSocket.accept(); //阻塞，等待连接的建立
HandlerThread handler = new HandlerThread();//开一个线程去处理该事件

引入多线程就能够很好的解决主线程被在处理网络请求的时候被阻塞的情况。随之而来引入的问题就是多线程上下文切换之间的开销，而且当请求很多的时候，线程较多会有c10k问题。所以就引入了IO多路复用的技术，使得主线程可以监听多个端口。

那么监听多个端口后好在哪呢？

考虑最普通的单线程同步socket：

while(true) {
    Socket socket = serverSocket.accept(); //阻塞
    processRequest(socket); //处理请求
}

如果processRequest(socket);时间较长，那么后面的到来的请求，要等到该函数处理完毕才可以和服务器建立连接。上面代码花时间的是处理网络请求，而不是在于连接的建立。所以我们将建立连接和处理网络请求剥离开，那么就能让服务器建立很多连接。进一步，在处理网络请求，我们就可以使用多线程来编程。而不需要使用每进入一个连接就开启一个线程。

NIO下的网络模型

在POSIX标准中,select(),poll(),epoll()都是能够实现io多路复用的api.虽然他们有些差异.不过提供的功能都是一样的.

调用select()的时候会让主线程阻塞,主线程来监听多个文件描述符,当其中某个描述符可以执行IO操作的时候,

select()结束阻塞,接着去执行io操作.

在java中,对于这样的操作做了封装,也就是Selector对象.各个Channel类都有register()函数,来将io事件注册到Selector中.

要使用Selector对象,首先是需要创建对象,代码如下,在将来,我们会使用register()函数将我们要关注的io事件注册到selector中;

Selector selector = Selector.open();

使用ServerSocketChannel

下面讲述NIO中的网络模型.我们实现一个Echoserver,就是说客户端发送什么,服务端返回什么.

为了能够接受http请求,我们首先需要为每一个端口多创建ServerSocketChannel对象.并且将其设置为非阻塞的.

for (int i=0; i<ports.length; ++i) {
    ServerSocketChannel ssc = ServerSocketChannel.open();
    ssc.configureBlocking( false );
    ServerSocket ss = ssc.socket();
    InetSocketAddress address = new InetSocketAddress( ports[i] ); //ip地址和端口
    ss.bind( address );  //将ServerSocket与ip地址绑定起来

    //将socket accept()事件注册到Selector
    SelectionKey key = ssc.register( selector, SelectionKey.OP_ACCEPT );

    System.out.println( "Going to listen on "+ports[i] );
}

等待IO请求

到目前为止,我们将我们要关注的io事件,即OP_ACCEPT,socket的建立加入到了Selector中.接下来只要等待IO事件的到来即可.

我们用一个死循环来永久地处理网络请求,在循环内部,我们调用selector.select()函数来阻塞当前线程,等待注册到Selector中的事件发生.当IO事件发生后selector.select()会不在阻塞,接着去执行下面的代码.

int num = selector.select(); //没有事件到来的时候会阻塞

Set selectedKeys = selector.selectedKeys();
Iterator it = selectedKeys.iterator();

while (it.hasNext()) {
    //..
}

响应io请求

当io事件到来后,我们要判断是何种类型的操作.上面代码Set selectedKeys = selector.selectedKeys();返回的是所有我们在监听的io事件.所以使用Iterator来遍历所有事件,判断事件的类型来执行我们对应的操作.

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

if ((key.readyOps() & SelectionKey.OP_ACCEPT) == SelectionKey.OP_ACCEPT) {
    // Accept the new connection
    ServerSocketChannel ssc = (ServerSocketChannel)key.channel();
    SocketChannel sc = ssc.accept();
    sc.configureBlocking( false );

    // Add the new connection to the selector
    SelectionKey newKey = sc.register( selector, SelectionKey.OP_READ );
    it.remove();

    System.out.println( "Got connection from "+sc );

如果当前到达的io事件是accept一个网络链接建立的请求.那么我们就去使用accept()来建立tcp链接.

注意,在前面的sc.configureBlocking( false );中我们配置了是非阻塞的网络请求,所以此时的accept并不会阻塞.会立刻建立网络链接.再加上此时到达的io事件是建立网络链接,一切都是那么地顺理成章,我们成功网络链接建立了.

接着我们要从客户端发送过来的数据进行读取,所以又往selector注册了read事件.

    SelectionKey newKey = sc.register( selector, SelectionKey.OP_READ );//注册read事件
    it.remove(); //当前事件被处理过后,就要从事件列表中移除

从客户端读取数据

上面的代码中,我们注册了从客户端读的事件.所以接下来就是判断io事件发生的时候,判断它是不是读的.并且做出对应的操作.总体逻辑和上面相类似,代码如下:

else if ((key.readyOps() & SelectionKey.OP_READ) == SelectionKey.OP_READ) {
    // Read the data
    SocketChannel sc = (SocketChannel)key.channel();
    // Echo data
    int bytesEchoed = 0;
    while (true) {
        echoBuffer.clear(); //每次重新读之前,清空缓冲区
        int r = sc.read( echoBuffer );
        if (r<=0) {
            break;
        }
        echoBuffer.flip(); //接下来我们要往SocketChannel中写入数据,所以先要flip()
        sc.write( echoBuffer );
        //Thread.sleep(1000*5);
        bytesEchoed += r;
    }

    System.out.println( "Echoed "+bytesEchoed+" from "+sc );
    it.remove(); //事件处理完毕后,从selector中移除
}

PS:

io多路复用是否意味着不需要使用多线程编程了?

当然不是的,如果将上面代码中的hread.sleep(1000*5);取消注释了.就会发现,主线程被阻塞的时候,其他端口也无法相应网络请求.这让io多路复用看起来十分鸡肋.因为主线程的阻塞还是会导致后续的操作无法继续下。

不过，我认为io多路复用的好处在于：可以监听多个channel是否可用吧。

The advantage of IoMultiplexing is that it allows blocking on multiple resources simultaneously, without needing to use polling (which wastes CPU cycles) or multithreading (which can be difficult to deal with, especially if threads are introduced into an otherwise sequential app only for the purpose of pending on multiple descriptors). -- https://wiki.c2.com/?IoMultiplexing

Summary

本文简要的介绍了NIO的基本操作.NIO和传统的流式IO(stream)相比来说,就是它在读取数据的时候是以块为单位的,而不是以字节流的形式,因此效率更高.NIO主要的操作对象是Channel和buffer,数据从外部读取到buffer中,如果要将数据写入到外部,也是先写入到buffer,在写入到channel中.

NIO也有全新的网络IO模型,实现了io多路复用的功能.在有些场景中,还需要TCP的half-open.nio中也提供了这样的api,使用shutdown可以实现该效果.不过这并不是本文的主要目的.就不在赘述