Java NIO入门

Java Java NIO

创建时间:2020-04-11 02:20

字数:3.9k 阅读:

NIO简介

Java NIO（New IO）是从java1.4之后引入的一个新的IO API，支持面向缓冲区的、基于通道的IO操作，以更高效的方式完成文件的读写。（Non Blocking IO）

传统IO（BIO）：类似于水流的字节流，单向的

NIO：类似于铁路，缓冲区=火车，双向流动，故称之为面向缓冲区

NIO的核心在于：通道（channel）和缓冲区（buffer），channel负责传输，buffer负责存储。

通道表示打开到IO设备的连接，使用NIO操作时要：获取用于连接IO设备的通道以及用于容纳数据的缓冲区。获取通道和缓冲区后，才可进行操作。

IO与NIO的主要区别：

IO	NIO
面向流	面向缓冲区
阻塞IO	非阻塞IO
无	选择器（Selectors）

缓冲区Buffer

缓冲区（Buffer）:

在 Java NIO 中负责数据的存取。缓冲区就是数组。用于存储不同数据类型的数据

根据数据类型的不同，Java提供了相应类型的缓冲区（boolean除外）：

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

上述缓冲区的管理方式几乎一致，通过allocate() 获取缓冲区
存取数据的核心方法：
- put() : 存入数据到缓冲区中
- get() : 取出缓冲区的数据
缓冲区的核心属性：

capacity：容量，表示缓冲区中最大存储的数据容量，一旦声明，不能改变。

limit：界限，表示缓冲区中可以操作数据的边界。（limit后的数据不能进行读写）

position：位置，缓冲区中正在操作数据的位置。

规则: 0 <= mark <= position <= limit <= capacity

Buffer API

下面代码以ByteBuffer为例子，列举出Buffer常用的方法：

    @Test
    public void test1(){
        String str="abcde";
        
        //allocate(int capacity)分配指定大小的缓冲区
        ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
        System.out.println("__________allocate()_________");
        System.out.println("position:"+byteBuffer.position());
        System.out.println("limit:\t"+byteBuffer.limit());
        System.out.println("capacity:"+byteBuffer.capacity());

        //利用put()将数组存入数据到缓冲区
        byteBuffer.put(str.getBytes());
        System.out.println("____________put()____________");
        System.out.println("position:"+byteBuffer.position());
        System.out.println("limit:\t"+byteBuffer.limit());
        System.out.println("capacity:"+byteBuffer.capacity());

        //flip()切换到读取数据的模式
        byteBuffer.flip();
        System.out.println("____________flip()___________");
        System.out.println("position:"+byteBuffer.position());
        System.out.println("limit:\t"+byteBuffer.limit());
        System.out.println("capacity:"+byteBuffer.capacity());

        //利用get()方法读取缓冲区中的数据
        byte[] dst = new byte[byteBuffer.limit()];
        byteBuffer.get(dst);
        System.out.println(new String(dst,0,dst.length));
        System.out.println("____________get()____________");
        System.out.println("position:"+byteBuffer.position());
        System.out.println("limit:\t"+byteBuffer.limit());
        System.out.println("capacity:"+byteBuffer.capacity());
        //rewind()重复读取数据
        byteBuffer.rewind();
        System.out.println("___________rewind()__________");
        System.out.println("position:"+byteBuffer.position());
        System.out.println("limit:\t"+byteBuffer.limit());
        System.out.println("capacity:"+byteBuffer.capacity());

        //clear()清空缓冲区,但缓冲区里面的数据依然存在，处于“被遗忘”状态
        byteBuffer.clear();
        System.out.println("___________clear()___________");
        System.out.println("position:"+byteBuffer.position());
        System.out.println("limit:\t"+byteBuffer.limit());
        System.out.println("capacity:"+byteBuffer.capacity());
    }

    @Test
    public void test2(){
        String str="abcde";
        ByteBuffer buf=ByteBuffer.allocate(1024);
        buf.put(str.getBytes());
        buf.flip();
        byte[] dst=new byte[buf.limit()];
        buf.get(dst,0,2);
        System.out.println(new String(dst,0,2));
        System.out.println("buf current position:"+buf.position());

        //mark() 标记当前position，与reset()搭配使用
        buf.mark();

        buf.get(dst,2,2);
        System.out.println(new String(dst,2,2));
        System.out.println("buf current position:"+buf.position());

        //reset() 恢复到mark时的位置
        buf.reset();
        System.out.println("buf current position:"+buf.position());

        //判断缓冲区中是否还有剩余的数据
        if (buf.hasRemaining()){
            //获取缓冲区中可以操作的数量
            System.out.println(buf.remaining());
        }
    }

直接缓冲区与非直接缓冲区

非直接缓冲区：通过allocate() 方法分配，缓冲区建立在JVM堆内存中
直接缓冲区：通过allocateDirect() 方法分配直接缓冲区，直接将缓冲区建立在操作系统主物理内存中，0拷贝，效率更高。通过垃圾回收释放

不足在于分配、销毁消耗大、不易控制
字节缓冲区要么是直接的，要么是非直接的。如果为直接字节缓冲区，则Java 虚拟机会尽最大努力直接在此缓冲区上执行本机I/O 操作。也就是说，在每次调用基础操作系统的一个本机I/O 操作之前（或之后），虚拟机都会尽量避免将缓冲区的内容复制到中间缓冲区中（或从中间缓冲区中复制内容）。
直接字节缓冲区可以通过调用此类的allocateDirect() 工厂方法来创建。此方法返回的缓冲区进行分配和取消分配所需成本通常高于非直接缓冲区。直接缓冲区的内容可以驻留在常规的垃圾回收堆之外，因此，它们对应用程序的内存需求量造成的影响可能并不明显。所以，**建议将直接缓冲区主要分配给那些易受基础系统的本机I/O 操作影响的大型、持久的缓冲区。**一般情况下，最好仅在直接缓冲区能在程序性能方面带来明显好处时分配它们。
直接字节缓冲区还可以通过FileChannel 的map() 方法将文件区域直接映射到内存中来创建。该方法返回MappedByteBuffer。Java 平台的实现有助于通过JNI 从本机代码创建直接字节缓冲区。如果以上这些缓冲区中的某个缓冲区实例指的是不可访问的内存区域，则试图访问该区域不会更改该缓冲区的内容，并且将会在访问期间或稍后的某个时间导致抛出不确定的异常。
字节缓冲区是直接缓冲区还是非直接缓冲区可通过调用其isDirect()方法来确定。提供此方法是为了能够在性能关键型代码中执行显式缓冲区管理。

左为OS，右为JVM

通道Channel

由java.nio.channels包定义的。channel表示IO源与目标打开的连接。Channel类似于传统的“流”，只不过channel本身不能直接访问数据，只能与buffer进行交互。

java.nio.channels.Channel为通道必须实现的接口，通道的主要实现类有：

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

获取通道的方式：

Java针对支持通道的类提供了getChannel() 方法：

本地IO：FileInputStream / FileOutputStream / RandomAccessFile

网络IO：Socket / ServerSocket / DatagramSocket
在JDK7中的NIO2 针对各个通道提供了静态方法 open()
在JDK7中的NIO2 的Files工具类的newByteChannel()

DEMO:

使用Buffer+通道完成文件复制：

@Test
public void test1() throws IOException {
    FileInputStream fis=new FileInputStream("1.jpg");
    FileOutputStream fos=new FileOutputStream("2.jpg");
    FileChannel in = fis.getChannel();
    FileChannel out = fos.getChannel();
    
    ByteBuffer buf=ByteBuffer.allocate(1024);
    while (in.read(buf)!=-1){
        buf.flip();
        out.write(buf);
        buf.clear();
    }
    
    out.close();
    in.close();
    fos.close();
    fis.close();
}

使用直接缓冲区完成文件的复制（内存映射文件）

@Test
public void test2() throws IOException {
    FileChannel inChannel=
        FileChannel.open(Paths.get("1.jpg"),StandardOpenOption.READ);
    FileChannel outChannel=
        FileChannel.open(Paths.get("3.jpg"),StandardOpenOption.WRITE,
                         StandardOpenOption.READ,StandardOpenOption.CREATE_NEW);
    //内存映射文件
    MappedByteBuffer inMapBuf = inChannel.map(FileChannel.MapMode.READ_ONLY, 0, inChannel.size());
    MappedByteBuffer outMapBuf = outChannel.map(FileChannel.MapMode.READ_WRITE, 0, inChannel.size());

    //直接对缓冲区进行数据的读写操作
    byte[] dst=new byte[inMapBuf.limit()];
    inMapBuf.get(dst);
    outMapBuf.put(dst);

    inChannel.close();
    outChannel.close();
}

使用直接缓冲区完成文件的复制（通道之间直接传输数据）

/**
 * 通道之间的数据传输（直接缓冲区）
 */
@Test
public void test3() throws IOException {
    FileChannel inChannel=FileChannel.open(Paths.get("1.jpg"),StandardOpenOption.READ);
    FileChannel outChannel=FileChannel.open(Paths.get("3.jpg"),StandardOpenOption.WRITE,
            StandardOpenOption.READ,StandardOpenOption.CREATE_NEW);

    inChannel.transferTo(0,inChannel.size(),outChannel);
    outChannel.transferFrom(inChannel,0,inChannel.size());

    inChannel.close();
    outChannel.close();
}

分散读取，聚集写入

分散读取（scattering reads）：从channel中读取的数据分散到多个buffer中（buffer数组），按照缓冲区的顺序，从channel中读取的数据依次将buffer填满

聚集写入（gathering write）：将多个Buffer 中的数据“聚集”到Channel：按照缓冲区的顺序，写入position 和limit 之间的数据到Channel 。

@Test
public void test4() throws IOException {
    RandomAccessFile raf1=new RandomAccessFile("hosts.txt","rw");
    //1.获取通道
    FileChannel channel1=raf1.getChannel();

    //2.分配指定大小的缓冲区
    ByteBuffer buf1 = ByteBuffer.allocate(100);
    ByteBuffer buf2 = ByteBuffer.allocate(1024);

    //3.分散读取：
    ByteBuffer[] buffers=new ByteBuffer[]{buf1,buf2};
    channel1.read(buffers);

    for (ByteBuffer buffer : buffers) {
        buffer.flip();
    }

    //打印各分散buffer详情
    System.out.println(new String(buffers[0].array(),0,buffers[0].limit()));
    System.out.println("___________________");
    System.out.println(new String(buffers[1].array(),0,buffers[1].limit()));

    //4.聚集写入
    RandomAccessFile raf2=new RandomAccessFile("2.txt","rw");
    FileChannel channel2 = raf2.getChannel();
    channel2.write(buffers);

    channel2.close();
}

字符集

编码：字符串 -》字节数组
解码：字节数组 -》字符串

在Java中，可以以Charset.forName("")的方法获取字符集对象，并进行针对特定字符集的编码解码操作：

@Test
public void test5() throws CharacterCodingException {
    Charset gbk = Charset.forName("GBK");
    CharsetEncoder encoder = gbk.newEncoder();  //  编码
    CharsetDecoder decoder = gbk.newDecoder();  //  解码

    CharBuffer buf=CharBuffer.allocate(1024);
    buf.put("学习使我快乐");
    buf.flip();

    //编码
    ByteBuffer byteBuffer=encoder.encode(buf);
    for (int i = 0; i < 12; i++) {  //Java中一个Char占两个字节
        System.out.println(byteBuffer.get());
    }

    byteBuffer.flip();	//读取byteBuffer
    CharBuffer charBuffer=decoder.decode(byteBuffer);
    System.out.println(charBuffer.toString());
}

NIO, 阻塞&非阻塞

传统的IO 流都是阻塞式的。也就是说，当一个线程调用read() 或write() 时，该线程被阻塞，直到有一些数据被读取或写入，该线程在此期间不能执行其他任务。因此，在完成网络通信进行IO 操作时，由于线程会阻塞，所以服务器端必须为每个客户端都提供一个独立的线程进行处理，当服务器端需要处理大量客户端时，性能急剧下降。
Java NIO 是非阻塞模式的。当线程从某通道进行读写数据时，若没有数据可用时，该线程可以进行其他任务。线程通常将非阻塞IO 的空闲时间用于在其他通道上执行IO 操作，所以单独的线程可以管理多个输入和输出通道。因此，NIO 可以让服务器端使用一个或有限几个线程来同时处理连接到服务器端的所有客户端。

使用NIO完成网络通信的核心：

通道Channel：负责连接。

常用通道一般继承自：SelectableChannel 抽象类，该类实现了：java.nio.channels.Channel 接口

SelectableChannel的子类有：

SocketChannel, ServerSocketChannel, DatagramChannel, Pipe.SinkChannel, Pipe.SourceChannel 等
缓冲区（Buffer）：

负责数据的存取
选择器（Selector）：

是SelectableChannel的多路复用器。用于监控SelectableChannel的IO状况。

阻塞式NIO

public class TestBlockingNIO {
    /**
     *客户端
     */
    @Test
    public void client() throws IOException{
        //1.获取通道
        SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress("127.0.0.1", 9898));
        FileChannel fileChannel = FileChannel.open(Paths.get("1.jpg"), StandardOpenOption.READ);

        //分配指定大小的缓冲区
        ByteBuffer buf=ByteBuffer.allocate(1024);

        //3.读取本地文件并发送到服务端
        while (fileChannel.read(buf)!=-1){
            buf.flip();
            socketChannel.write(buf);
            buf.clear();
        }

        //4.关闭通道
        fileChannel.close();
        socketChannel.close();
    }

    /**
     *服务器
     */
    @Test
    public void server() throws IOException {
        //1.获取通道
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        FileChannel fileChannel = FileChannel.open(Paths.get("2.jpg"), StandardOpenOption.WRITE, StandardOpenOption.CREATE);
        //2.绑定端口
        serverSocketChannel.bind(new InetSocketAddress(9898));
        //3.获取客户端连接的通道
        System.out.println("等待client连接...");
        SocketChannel accept = serverSocketChannel.accept();
        System.out.println("Client连接成功...:"+accept.getRemoteAddress()+"|"+accept.getLocalAddress());
        //4.分配指定大小的缓冲区
        ByteBuffer buffer=ByteBuffer.allocate(1024);
        //5.接收客户端传来的数据并保存到本地
        while (accept.read(buffer)!=-1){
            buffer.flip();
            fileChannel.write(buffer);
            buffer.clear();
        }
        //6.关闭通道
        System.out.println("关闭server...");
        accept.close();
        fileChannel.close();
        serverSocketChannel.close();
    }
}

非阻塞式NIO

¶选择器（Selector）：

是SelectableChannle 对象的多路复用器，Selector 可以同时监控多个SelectableChannel 的IO 状况，也就是说，利用Selector 可使一个单独的线程管理多个Channel。Selector 是非阻塞IO 的核心
创建：

通过调用Selector.open() 方法创建一个Selector
```
Selector selector = Selector.open();
```
向选择器注册一个channel：
当调用register(Selector sel, int ops) 将通道注册选择器时，选择器对通道的监听事件，需要通过第二个参数ops 指定。
可以监听的事件类型（可使用SelectionKey 的四个常量表示）：
- 读: SelectionKey.OP_READ （1）
- 写: SelectionKey.OP_WRITE （4）
- 连接: SelectionKey.OP_CONNECT（8）
- 接收: SelectionKey.OP_ACCEPT （16）
若注册时不止监听一个事件，则可以使用“位或”操作符连接
Selector常用方法：

¶SelectionKey

SelectionKey：

表示SelectableChannel 和Selector 之间的注册关系。每次向选择器注册通道时就会选择一个事件(选择键)。选择键包含两个表示为整数值的操作集。操作集的每一位都表示该键的通道所支持的一类可选择操作。
SelectionKey常用方法：

¶TCP DEMO

public class TestNoBlockingNIO {
    
    @Test
    public void client() throws IOException {
        //1.获取通道
        SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress("127.0.0.1", 9898));
        //2.切换为非阻塞模式
        socketChannel.configureBlocking(false);
        //3.分配指定大小的缓冲区
        ByteBuffer buf=ByteBuffer.allocate(1024);

        //4.发送数据给server
        Scanner scanner=new Scanner(System.in);
        while (scanner.hasNext()){
            String str = scanner.next();
            buf.put((new Date().toString()+"\n"+str).getBytes());
            buf.flip();
            socketChannel.write(buf);
            buf.clear();
        }


        //4.关闭通道
        socketChannel.close();
    }
    
    
    @Test
    public void server() 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> iterator = selector.selectedKeys().iterator();
            while (iterator.hasNext()){
                //8. 获取准备就绪的事件
                SelectionKey sk = iterator.next();
                //9. 判断具体是什么事件准备就绪
                if (sk.isAcceptable()){
                    //10. 若“接收就绪”，获取客户端连接
                    SocketChannel accept = ssChannel.accept();
                    //11. 切换为非阻塞模式
                    accept.configureBlocking(false);
                    //12. 将该通道注册到选择器上
                    accept.register(selector,SelectionKey.OP_READ);
                }else if (sk.isReadable()){
                    //13. 获取当前选择器上“读就绪”的通道
                    SocketChannel channel = (SocketChannel) sk.channel();
                    //14. 读取数据
                    ByteBuffer buffer=ByteBuffer.allocate(1024);
                    int len=0;
                    while ((len=channel.read(buffer))>0){
                        buffer.flip();
                        System.out.println(new String(buffer.array(),0,len));
                        buffer.clear();
                    }
                }
                //15.记得取消选择键SelectionKey
                iterator.remove();
            }

        }
    }
}

测试效果：

Clients：

Server:

另外：

已就绪的选择键，可根据类型分给不同的线程去处理

¶UDP通信的例子

public class TestNoBlockingNIO2 {
    @Test
    public void send() throws IOException {
        InetSocketAddress server_addr = new InetSocketAddress("127.0.0.1", 9898);
        //1.获取通道
        DatagramChannel dc = DatagramChannel.open();
        //2.切换为非阻塞模式
        dc.configureBlocking(false);
        //3.分配指定大小的缓冲区
        ByteBuffer buf=ByteBuffer.allocate(1024);

        //4.发送数据给server
        Scanner scanner=new Scanner(System.in);
        while (scanner.hasNext()){
            String str = scanner.next();
            buf.put((new Date().toString()+"\n"+str).getBytes());
            buf.flip();
            dc.send(buf,server_addr);
            buf.clear();
        }


        //4.关闭通道
        dc.close();
    }
    @Test
    public void receive() 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> iterator = selector.selectedKeys().iterator();
            while (iterator.hasNext()){
                SelectionKey sk = iterator.next();
                if (sk.isReadable()){
                    ByteBuffer buffer=ByteBuffer.allocate(1024);
                    dc.receive(buffer);
                    buffer.flip();
                    System.out.println(new String(buffer.array(),0,buffer.limit()));
                    buffer.clear();
                }
                iterator.remove();
            }
        }
    }
}