Netty 专栏——NIO

一、Buffer

1、基本概念

Buffer 的实质就是一个数组或容器对象，是可以读写的内存块，Buffer对象提供了一组操作方法，用于对内存块进行操作，Buffer 内置了一些机制，能够跟踪和记录缓冲区的状态变化情况。

2、Buffer的常用子类

Buffer的常用子类有：

• 1）ByteBuffer：存储字节数据到缓冲区
• 2）StringBuffer：存储字符串数据到缓冲区
• 3）CharBuffer：存储字符数据到缓冲区
• 4）IntBuffer：存储整数数据到缓冲区
• 5）LongBuffer：存储长整型数据到缓冲区
• 6）DoubleBuffer：存储双精度实数到缓冲区
• 7）FloatBuffer：存储小数到缓冲区

3、Buffer的分散和聚集

分散：Scattering，将数据写入到buffer时，采用buffer数组，依次写入；
聚集：Gathering，从buffer读取数据时，采用buffer数组，依次读取

/* Scattering: using buffer array to write data [scatter] Gathering: read data from buffer array * */
public static void scatterAndGatherDemo() throws Exception{
    
    /*using server socket channel and socket channel*/
    ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
    /*create server address*/
    InetSocketAddress socketAddress = new InetSocketAddress(8001);
    /*binding port into server socket*/
    serverSocketChannel.socket().bind(socketAddress);
    /*create buffer array*/
    ByteBuffer[] byteBuffers = new ByteBuffer[2];
    byteBuffers[0] = ByteBuffer.allocate(5);
    byteBuffers[1] = ByteBuffer.allocate(3);
    /*wait client to connect*/
    SocketChannel socketChannel = serverSocketChannel.accept();
    /*read data through while*/
    int messageLength = 8;
    while(true){
    
        int byteRead = 0;
        while(byteRead < messageLength){
    
            long read = socketChannel.read(byteBuffers);
            byteRead += read;
            System.out.println("byteRead=" + byteRead);
            /*using stream print to see the position and limit of buffer*/
            Arrays.asList(byteBuffers).stream().map(buffer -> "position=" + buffer.position() +
                                                    ", limit=" + buffer.limit()).forEach(System.out::println);

        }
        /*transform all buffer*/
        Arrays.asList(byteBuffers).forEach(buffer -> buffer.flip());

        /*show data to client*/
        long byteWrite = 0;
        while(byteWrite < messageLength){
    
            long write = socketChannel.write(byteBuffers);
            byteWrite += write;
        }
        /*clear all buffer*/
        Arrays.asList(byteBuffers).forEach(buffer ->{
    
            buffer.clear();
        });
        System.out.println("byteWrite=" + byteWrite + ", byteRead=" + byteRead +
                           ", message length="+messageLength);
    }
}

二、Channel

1、基本概念

Channel 类似于流，但是Channel 能够同时进行读写，而流只能读或写；Channel 可以异步读写数据，可以读写到缓冲区中。

2、常用子类

Channel的常用子类有FileChannel、DatagramChannel、ServerSocketChannel和SocketChannel

• 1）FileChannel：用于文件数据的读写
• 2）DatagramChannel：用于UDP数据的读写
• 3）ServerSocketChannel 和 SockeChannel：用于 TCP 数据的读写

3、FileChannel示例

写数据到文件中

public class FileChannelDemo {
    
    public static void main(String[] args) throws Exception{
    
        /*create a string data*/
        String str = "Hello, channel of nio，你好，同步非阻塞的文件通道！";
        /*create a output stream*/
        FileOutputStream out = new FileOutputStream("./file.txt");
        /*through the output stream to attain a file channel*/
        FileChannel fileChannel = out.getChannel();
        /*create a buffer*/
        ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
        /*write the str into the buffer*/
        byteBuffer.put(str.getBytes(StandardCharsets.UTF_8));
        /*file channel read data from buffer*/
        byteBuffer.flip();
        fileChannel.write(byteBuffer);
        out.close();
    }
}

从文件中读数据并打印到控制台中

/*create file file handle*/
File file = new File("./file.txt");
/*create a input stream*/
FileInputStream in = new FileInputStream(file);
/*create channel*/
FileChannel channel = in.getChannel();
/*create byte buffer*/
ByteBuffer byteBuffer = ByteBuffer.allocate((int) file.length());
/*read data form channel into buffer*/
channel.read(byteBuffer);
/*show data in console*/
String str = new String(byteBuffer.array());
System.out.println(str);
in.close();
拷贝文件,使用两个fileChannel
public static void copyFile() throws Exception{
    
    /*create file file handle*/
    File file = new File("./file.txt");
    /*create a input stream*/
    FileInputStream in = new FileInputStream(file);
    /*create input channel*/
    FileChannel channel = in.getChannel();
    /*create byte buffer*/
    ByteBuffer byteBuffer = ByteBuffer.allocate((int) file.length());
    /*read data form channel into buffer*/
    channel.read(byteBuffer);
    /*create out put stream*/
    FileOutputStream out = new FileOutputStream("./file01.txt");
    /*create out channel*/
    FileChannel outChannel = out.getChannel();
    /*write data from buffer to out channel*/
    byteBuffer.flip();
    outChannel.write(byteBuffer);
    out.close();
    in.close();
}

使用transform完成拷贝

public static void transformDemo() throws Exception{
    

    /*create a input stream, read the image data*/
    FileInputStream fileInputStream = new FileInputStream("D:\\图片\\Saved Pictures\\200.jpg");
    /*create a output stream*/
    FileOutputStream fileOutputStream = new FileOutputStream("./200-bake.jpg");
    /*get the file channel*/
    FileChannel inChannel = fileInputStream.getChannel();
    FileChannel outChannel = fileOutputStream.getChannel();
    /*using the transform method to finish the data copy*/
    outChannel.transferFrom(inChannel,0,inChannel.size());
    /*close channel and stream*/
    inChannel.close();
    outChannel.close();
    fileInputStream.close();
    fileOutputStream.close();

}

4、Channel 和 Buffer 的小总结

• 1）Buffer 支持类型化的 get 和 put，放入的是什么样类型的数据，读取的时候就应该用什么样的类型变量去接收，不然将会出现类型不支持异常BufferUnderflowException；

public static void demo02(){
    
    ByteBuffer byteBuffer = ByteBuffer.allocate(20);
    byteBuffer.putInt(100);
    byteBuffer.putLong(20);
    byteBuffer.putChar('御');
    byteBuffer.putShort((short) 14);
    byteBuffer.flip();
    System.out.println(byteBuffer.getInt());
    System.out.println(byteBuffer.getLong());
    System.out.println(byteBuffer.getChar());
    System.out.println(byteBuffer.getShort());
}

• 2）通过 Buffer 的 asReadOnlyBuffer 方法将一个普通的 Buffer 转换成只读buffer

/** * Creates a new, read-only byte buffer that shares this buffer's * content. * * <p> The content of the new buffer will be that of this buffer. Changes * to this buffer's content will be visible in the new buffer; the new * buffer itself, however, will be read-only and will not allow the shared * content to be modified. The two buffers' position, limit, and mark * values will be independent. * * <p> The new buffer's capacity, limit, position, and mark values will be * identical to those of this buffer. * * <p> If this buffer is itself read-only then this method behaves in * exactly the same way as the {@link #duplicate duplicate} method. </p> * * @return The new, read-only byte buffer */
    public abstract ByteBuffer asReadOnlyBuffer();
    ```
- 3）NIO 提供了 MappedByteBuffer， 可以让文件直接在本地内存(堆外内存）中进行修改，然后同步到文件中的操作由 NIO 完成
```java
public static void mappedDemo() throws Exception{
    
    RandomAccessFile accessFile = new RandomAccessFile("./file01.txt", "rw");
    /*get channel*/
    FileChannel fileChannel = accessFile.getChannel();
    /*option*/
    /*parameter of map: first parameter used to set mode of read or write file * the second parameter is used to set the position of read or write in file * the third parameter is used to set the size of data of you want to read or write * */
    MappedByteBuffer map = fileChannel.map(FileChannel.MapMode.READ_WRITE, 0, 5);
    map.put(0,(byte) 'H');
    map.put(3,(byte) '9');
    accessFile.close();
}

• 4）NIO 支持通过多个 buffer ，即 Buffer 数组完成读写操作，即 Scattering 和 Gathering。

三、Selector

1、基本概念

Selector 允许多个Channel以事件的形式注册到Selector 上，并能够检测多个注册的通道上是否有事件发生，并根据对应的事件选择对应的处理方法进行处理；其特点是，只有当连接真正有读写事件发生时，才会进行读写，大大减小系统开销；

2、重要成员方法

• 1）open：得到一个选择器对象
• 2）select：监控所有注册的通道，当其中有事件发生时，选中并进行相应处理
• 3）selectedKeys：从内部集合中得到所有的selectionKey

细节：selelct() 如果不设置timeout参数，将表现为阻塞方法，设置参数后则超时返回，而selectNow() 则不阻塞立马返回，wakeup() 用于唤醒selector

3、原理

• 1）当客户端连接时，会通过ServerSocketChannel 得到 SocketChannel；
• 2）将 SocketChannel 注册到 Selector 上
• 3）注册后返回一个 SelectionKey（标识符），会和该 Selector 关联（集合）
• 4）Selector 通过 select 方法监听 Channel 中的事件：读事件、写事件、连接已创建事件（connect）和连接请求事件（accept），进一步得到各个有事件发生的 Channel的 SelectionKey
• 5）通过 SelectionKey 可以反向获取 SocketChannel
• 6）通过得到的 Channel 完成相应的事件处理

4、ServerDemo

public class ServerDemo {
    
    public static void main(String[] args) throws Exception{
    
        /*create a Server Socket Channel*/
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();

        /*get Selector instance*/
        Selector selector = Selector.open();

        /*binding port*/
        serverSocketChannel.socket().bind(new InetSocketAddress(8001));

        /*set non block*/
        serverSocketChannel.configureBlocking(false);

        /*registry the server socket channel into selector, focus on the OP_ACCEPT*/
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

        /*wait the client to connect*/
        while(true){
    
            /*if not event happen, continue after one second*/
            if(selector.select(1000) == 0){
    
                System.out.println("Having waited one second");
                continue;
            }
            /*if the return value >0, get the selection key set*/
            /* * if return value is larger than 0 that meaning that the selector have got event of concern * */
            Set<SelectionKey> selectionKeys = selector.selectedKeys();

            /*iterator the selection key set*/
            Iterator<SelectionKey> keyIterator = selectionKeys.iterator();
            while(keyIterator.hasNext()){
    
                /*get concrete selection key*/
                SelectionKey key = keyIterator.next();
                /*determine the concrete event of the key*/
                if(key.isAcceptable()){
    
                    /*if this event is a new connect request from client*/
                    /*create a new socket channel to the client*/
                    SocketChannel socketChannel = serverSocketChannel.accept();
                    socketChannel.configureBlocking(false);
                    /*registrying this socket channel to selector and binding a buffer*/
                    socketChannel.register(selector,SelectionKey.OP_READ,
                                           ByteBuffer.allocate(1024));
                }else if(key.isReadable()){
    
                    /*if this is a read event*/
                    /*get concrete channel*/
                    SocketChannel socketChannel = (SocketChannel) key.channel();
                    /*get the buffer of associating with the socket channel*/
                    ByteBuffer byteBuffer = (ByteBuffer) key.attachment();
                    socketChannel.read(byteBuffer);
                    System.out.println(" message form client: " + new String(byteBuffer.array()));
                }
                /*delete the concurrent selection key form selection key set, avoid repeat option*/
                keyIterator.remove();
            }
        }
    }
}

5、ClientDemo

public class ClientDemo {
    
    public static void main(String[] args) throws Exception{
    
        /*get socket Channel*/
        SocketChannel socketChannel = SocketChannel.open();
        /*set not blocking*/
        socketChannel.configureBlocking(false);
        /*support the ip and port of server side*/
        InetSocketAddress address = new InetSocketAddress("127.0.0.1", 8001);
        /*connect to server side*/
        if(!socketChannel.connect(address)){
    
            while (!socketChannel.finishConnect()){
    
                System.out.println("Connecting-----");
            }
        }
        /*if finished connect option*/
        String str = "Hello NIO Server,你好非阻塞";
        ByteBuffer byteBuffer = ByteBuffer.wrap(str.getBytes(StandardCharsets.UTF_8));
        /*send data*/
        socketChannel.write(byteBuffer);
        int read = System.in.read();
        socketChannel.close();
    }
}