Java核心知识点-NIO
博客专区 > Yemon 的博客 > 博客详情
Java核心知识点-NIO
Yemon 发表于2年前
Java核心知识点-NIO
  • 发表于 2年前
  • 阅读 6
  • 收藏 0
  • 点赞 2
  • 评论 0

腾讯云 新注册用户 域名抢购1元起>>>   

文件读取中的NIO

在Java1.4之前的I/O系统中,提供的都是面向流的I/O系统,系统一次一个字节地处理数据,一个输入流产生一个字节的数据,一个输出流消费一个字节的数据,面向流的I/O速度非常慢,而在Java 1.4中推出了NIO,这是一个面向块的I/O系统,系统以块的方式处理处理,每一个操作在一步中产生或者消费一个数据库,按块处理要比按字节处理数据快的多。

在NIO中有几个核心对象需要掌握:缓冲区(Buffer)、通道(Channel)、选择器(Selector)。

缓冲区Buffer

缓冲区实际上是一个容器对象,更直接的说,其实就是一个数组,在NIO库中,所有数据都是用缓冲区处理的。在读取数据时,它是直接读到缓冲区中的; 在写入数据时,它也是写入到缓冲区中的;任何时候访问 NIO 中的数据,都是将它放到缓冲区中。而在面向流I/O系统中,所有数据都是直接写入或者直接将数据读取到Stream对象中。

在NIO中,所有的缓冲区类型都继承于抽象类Buffer,最常用的就是ByteBuffer,对于Java中的基本类型,基本都有一个具体Buffer类型与之相对应,它们之间的继承关系如下图所示:

 

下面是一个简单的使用IntBuffer的例子:

import java.nio.IntBuffer;

public class TestIntBuffer {
	public static void main(String[] args) {
		// 分配新的int缓冲区,参数为缓冲区容量
		// 新缓冲区的当前位置将为零,其界限(限制位置)将为其容量。它将具有一个底层实现数组,其数组偏移量将为零。
		IntBuffer buffer = IntBuffer.allocate(8);

		for (int i = 0; i < buffer.capacity(); ++i) {
			int j = 2 * (i + 1);
			// 将给定整数写入此缓冲区的当前位置,当前位置递增
			buffer.put(j);
		}

		// 重设此缓冲区,将限制设置为当前位置,然后将当前位置设置为0
		buffer.flip();

		// 查看在当前位置和限制位置之间是否有元素
		while (buffer.hasRemaining()) {
			// 读取此缓冲区当前位置的整数,然后当前位置递增
			int j = buffer.get();
			System.out.print(j + " ");
		}

	}

}

运行后可以看到:

通道Channel

通道是一个对象,通过它可以读取和写入数据,当然了所有数据都通过Buffer对象来处理。我们永远不会将字节直接写入通道中,相反是将数据写入包含一个或者多个字节的缓冲区。同样不会直接从通道中读取字节,而是将数据从通道读入缓冲区,再从缓冲区获取这个字节。

在NIO中,提供了多种通道对象,而所有的通道对象都实现了Channel接口。它们之间的继承关系如下图所示:

使用NIO读取数据

在前面我们说过,任何时候读取数据,都不是直接从通道读取,而是从通道读取到缓冲区。所以使用NIO读取数据可以分为下面三个步骤:
1. 从FileInputStream获取Channel
2. 创建Buffer
3. 将数据从Channel读取到Buffer中

下面是一个简单的使用NIO从文件中读取数据的例子:

import java.io.*;
import java.nio.*;
import java.nio.channels.*;

public class Program {
    static public void main( String args[] ) throws Exception {
        FileInputStream fin = new FileInputStream("c:\\test.txt");
        
        // 获取通道
        FileChannel fc = fin.getChannel();
        
        // 创建缓冲区
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        
        // 读取数据到缓冲区
        fc.read(buffer);
        
        buffer.flip();
        
        while (buffer.remaining()>0) {
            byte b = buffer.get();
            System.out.print(((char)b));
        }
        
        fin.close();
    }
}

使用NIO写入数据

使用NIO写入数据与读取数据的过程类似,同样数据不是直接写入通道,而是写入缓冲区,可以分为下面三个步骤:
1. 从FileInputStream获取Channel
2. 创建Buffer
3. 将数据从Channel写入到Buffer中

下面是一个简单的使用NIO向文件中写入数据的例子:

import java.io.*;  
import java.nio.*;  
import java.nio.channels.*;  
  
public class Program {  
    static private final byte message[] = { 8311110910132,  
        9812111610111546 };  
  
    static public void main( String args[] ) throws Exception {  
        FileOutputStream fout = new FileOutputStream( "c:\\test.txt" );  
          
        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 );  
          
        fout.close();  
    }  
}  

在第一篇中,我们介绍了NIO中的两个核心对象:缓冲区和通道,在谈到缓冲区时,我们说缓冲区对象本质上是一个数组,但它其实是一个特殊的数组,缓冲区对象内置了一些机制,能够跟踪和记录缓冲区的状态变化情况,如果我们使用get()方法从缓冲区获取数据或者使用put()方法把数据写入缓冲区,都会引起缓冲区状态的变化。本文为NIO使用及原理分析的第二篇,将会分析NIO中的Buffer对象。

在缓冲区中,最重要的属性有下面三个,它们一起合作完成对缓冲区内部状态的变化跟踪:

position:指定了下一个将要被写入或者读取的元素索引,它的值由get()/put()方法自动更新,在新创建一个Buffer对象时,position被初始化为0。

limit:指定还有多少数据需要取出(在从缓冲区写入通道时),或者还有多少空间可以放入数据(在从通道读入缓冲区时)。

capacity:指定了可以存储在缓冲区中的最大数据容量,实际上,它指定了底层数组的大小,或者至少是指定了准许我们使用的底层数组的容量。

以上四个属性值之间有一些相对大小的关系:0 <= position <= limit <= capacity。如果我们创建一个新的容量大小为10的ByteBuffer对象,在初始化的时候,position设置为0,limit和 capacity被设置为10,在以后使用ByteBuffer对象过程中,capacity的值不会再发生变化,而其它两个个将会随着使用而变化。四个属性值分别如图所示:

 

现在我们可以从通道中读取一些数据到缓冲区中,注意从通道读取数据,相当于往缓冲区中写入数据。如果读取4个自己的数据,则此时position的值为4,即下一个将要被写入的字节索引为4,而limit仍然是10,如下图所示:

 

下一步把读取的数据写入到输出通道中,相当于从缓冲区中读取数据,在此之前,必须调用flip()方法,该方法将会完成两件事情:

1. 把limit设置为当前的position值
2. 把position设置为0

由于position被设置为0,所以可以保证在下一步输出时读取到的是缓冲区中的第一个字节,而limit被设置为当前的position,可以保证读取的数据正好是之前写入到缓冲区中的数据,如下图所示:

 

现在调用get()方法从缓冲区中读取数据写入到输出通道,这会导致position的增加而limit保持不变,但position不会超过limit的值,所以在读取我们之前写入到缓冲区中的4个自己之后,position和limit的值都为4,如下图所示:

 

在从缓冲区中读取数据完毕后,limit的值仍然保持在我们调用flip()方法时的值,调用clear()方法能够把所有的状态变化设置为初始化时的值,如下图所示:

 

最后我们用一段代码来验证这个过程,如下所示:

import java.io.*;
import java.nio.*;
import java.nio.channels.*;

public class Program {
	public static void main(String args[]) throws Exception {
		FileInputStream fin = new FileInputStream("d:\\test.txt");
		FileChannel fc = fin.getChannel();

		ByteBuffer buffer = ByteBuffer.allocate(10);
		output("初始化", buffer);

		fc.read(buffer);
		output("调用read()", buffer);

		buffer.flip();
		output("调用flip()", buffer);

		while (buffer.remaining() > 0) {
			byte b = buffer.get();
			// System.out.print(((char)b));
		}
		output("调用get()", buffer);

		buffer.clear();
		output("调用clear()", buffer);

		fin.close();
	}

	public static void output(String step, Buffer buffer) {
		System.out.println(step + " : ");
		System.out.print("capacity: " + buffer.capacity() + ", ");
		System.out.print("position: " + buffer.position() + ", ");
		System.out.println("limit: " + buffer.limit());
		System.out.println();
	}
}

完成的输出结果为:

在上一篇文章中介绍了缓冲区内部对于状态变化的跟踪机制,而对于NIO中缓冲区来说,还有很多的内容值的学习,如缓冲区的分片与数据共享,只读缓冲区等。在本文中我们来看一下缓冲区一些更细节的内容。

缓冲区的分配

在前面的几个例子中,我们已经看过了,在创建一个缓冲区对象时,会调用静态方法allocate()来指定缓冲区的容量,其实调用 allocate()相当于创建了一个指定大小的数组,并把它包装为缓冲区对象。或者我们也可以直接将一个现有的数组,包装为缓冲区对象,如下示例代码所示:

public class BufferWrap {

    public void myMethod() {
        // 分配指定大小的缓冲区
        ByteBuffer buffer1 = ByteBuffer.allocate(10);
        
        // 包装一个现有的数组
        byte array[] = new byte[10];
        ByteBuffer buffer2 = ByteBuffer.wrap( array );
    }
}

缓冲区分片

在NIO中,除了可以分配或者包装一个缓冲区对象外,还可以根据现有的缓冲区对象来创建一个子缓冲区,即在现有缓冲区上切出一片来作为一个新的缓冲区,但现有的缓冲区与创建的子缓冲区在底层数组层面上是数据共享的,也就是说,子缓冲区相当于是现有缓冲区的一个视图窗口。调用slice()方法可以创建一个子缓冲区,让我们通过例子来看一下:

import java.nio.*;

public class Program {
    static public void main( String args[] ) throws Exception {
        ByteBuffer buffer = ByteBuffer.allocate( 10 );
        
        // 缓冲区中的数据0-9
        for (int i=0; i<buffer.capacity(); ++i) {
            buffer.put( (byte)i );
        }
        
        // 创建子缓冲区
        buffer.position( 3 );
        buffer.limit( 7 );
        ByteBuffer slice = buffer.slice();
        
        // 改变子缓冲区的内容
        for (int i=0; i<slice.capacity(); ++i) {
            byte b = slice.get( i );
            b *= 10;
            slice.put( i, b );
        }
        
        buffer.position( 0 );
        buffer.limit( buffer.capacity() );
        
        while (buffer.remaining()>0) {
            System.out.println( buffer.get() );
        }
    }
}

在该示例中,分配了一个容量大小为10的缓冲区,并在其中放入了数据0-9,而在该缓冲区基础之上又创建了一个子缓冲区,并改变子缓冲区中的内容,从最后输出的结果来看,只有子缓冲区“可见的”那部分数据发生了变化,并且说明子缓冲区与原缓冲区是数据共享的,输出结果如下所示:

 

只读缓冲区

只读缓冲区非常简单,可以读取它们,但是不能向它们写入数据。可以通过调用缓冲区的asReadOnlyBuffer()方法,将任何常规缓冲区转 换为只读缓冲区,这个方法返回一个与原缓冲区完全相同的缓冲区,并与原缓冲区共享数据,只不过它是只读的。如果原缓冲区的内容发生了变化,只读缓冲区的内容也随之发生变化:

import java.nio.*;

public class Program {
    static public void main( String args[] ) throws Exception {
        ByteBuffer buffer = ByteBuffer.allocate( 10 );
        
        // 缓冲区中的数据0-9
        for (int i=0; i<buffer.capacity(); ++i) {
            buffer.put( (byte)i );
        }

        // 创建只读缓冲区
        ByteBuffer readonly = buffer.asReadOnlyBuffer();
        
        // 改变原缓冲区的内容
        for (int i=0; i<buffer.capacity(); ++i) {
            byte b = buffer.get( i );
            b *= 10;
            buffer.put( i, b );
        }
        
        readonly.position(0);
        readonly.limit(buffer.capacity());
        
        // 只读缓冲区的内容也随之改变
        while (readonly.remaining()>0) {
            System.out.println( readonly.get());
        }
    }
}

 

如果尝试修改只读缓冲区的内容,则会报ReadOnlyBufferException异常。只读缓冲区对于保护数据很有用。在将缓冲区传递给某个 对象的方法时,无法知道这个方法是否会修改缓冲区中的数据。创建一个只读的缓冲区可以保证该缓冲区不会被修改。只可以把常规缓冲区转换为只读缓冲区,而不能将只读的缓冲区转换为可写的缓冲区。

直接缓冲区

直接缓冲区是为加快I/O速度,使用一种特殊方式为其分配内存的缓冲区,JDK文档中的描述为:给定一个直接字节缓冲区,Java虚拟机将尽最大努 力直接对它执行本机I/O操作。也就是说,它会在每一次调用底层操作系统的本机I/O操作之前(或之后),尝试避免将缓冲区的内容拷贝到一个中间缓冲区中 或者从一个中间缓冲区中拷贝数据。要分配直接缓冲区,需要调用allocateDirect()方法,而不是allocate()方法,使用方式与普通缓冲区并无区别,如下面的拷贝文件示例:

import java.io.*;
import java.nio.*;
import java.nio.channels.*;

public class Program {
    static public void main( String args[] ) throws Exception {
        String infile = "c:\\test.txt";
        FileInputStream fin = new FileInputStream( infile );
        FileChannel fcin = fin.getChannel();
        
        String outfile = String.format("c:\\testcopy.txt");
        FileOutputStream fout = new FileOutputStream( outfile );    
        FileChannel fcout = fout.getChannel();
        
        // 使用allocateDirect,而不是allocate
        ByteBuffer buffer = ByteBuffer.allocateDirect( 1024 );
        
        while (true) {
            buffer.clear();
            
            int r = fcin.read( buffer );
            
            if (r==-1) {
                break;
            }
            
            buffer.flip();
            
            fcout.write( buffer );
        }
    }
}

内存映射文件I/O

内存映射文件I/O是一种读和写文件数据的方法,它可以比常规的基于流或者基于通道的I/O快的多。内存映射文件I/O是通过使文件中的数据出现为 内存数组的内容来完成的,这其初听起来似乎不过就是将整个文件读到内存中,但是事实上并不是这样。一般来说,只有文件中实际读取或者写入的部分才会映射到内存中。如下面的示例代码:

import java.io.*;
import java.nio.*;
import java.nio.channels.*;

public class Program {
    static private final int start = 0;<span style="font-family:FangSong_GB2312;font-size:13px;">
    static private final int size = 1024;
    
    static public void main( String args[] ) throws Exception {
        RandomAccessFile raf = new RandomAccessFile( "c:\\test.txt", "rw" );
        FileChannel fc = raf.getChannel();
        
        MappedByteBuffer mbb = fc.map( FileChannel.MapMode.READ_WRITE,
          start, size );
        
        mbb.put( 0, (byte)97 );
        mbb.put( 1023, (byte)122 );
        
        raf.close();
    }
}

 

关于缓冲区的细节内容,我们已经用了两篇文章来介绍。在下一篇中将会介绍NIO中更有趣的部分Nonblocking I/O。

 

网络传输中的NIO

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

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

  • 事件驱动机制:事件到的时候触发,而不是同步的去监视事件。

  • 线程通讯:线程之间通过wait,notify等方式通讯。保证每次上下文切换都是有意义的。减少无谓的线程切换。(//核心所在)

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

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

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

事件名                                                 对应值                                                         
服务端接收客户端连接事件 SelectionKey.OP_ACCEPT(16)
客户端连接服务端事件 SelectionKey.OP_CONNECT(8)
读事件 SelectionKey.OP_READ(1)
写事件 SelectionKey.OP_WRITE(4)

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

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

服务端

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       */  
    
//这里其实相当于一个单线程循环询问是否有消息到达,如果有就各种处理。

//这里selector.select();目的就是减少上下文切换,内部类似wait()等待,等待的线程不会被进程调度程序调到cpu上运行,如果有消息到了,就notify()通知一下。

//简单比喻,你点了外卖就等着wait(),如果你的外卖到就会通知notify()你去拿外卖。

    @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();  
    }  
  
}  

客户端

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();  
    }  
  
}

 

共有 人打赏支持
粉丝 11
博文 320
码字总数 22799
×
Yemon
如果觉得我的文章对您有用,请随意打赏。您的支持将鼓励我继续创作!
* 金额(元)
¥1 ¥5 ¥10 ¥20 其他金额
打赏人
留言
* 支付类型
微信扫码支付
打赏金额:
已支付成功
打赏金额: