NIO学习笔记

忘是亡心i 2022-12-30 08:05 135阅读 0赞

### NIO ###

操作系统背景知识

> unix提供了5中io模型，其中java的底层实现依赖的是操作系统的**io复用模型**。linux提供select/poll，进程通过将一个或多个fd(文件描述符)传递给select或者poll，阻塞在select上，这样select/poll可以帮我们侦测多个fd是否处于就绪状态。select/poll顺序扫描fd是否就绪，并且支持的fd数量有限。linux还提供了一个epoll系统调用，使用事件驱动的方式代替顺序扫描fd，当有fd就绪时，执行回调函数rollback，因此性能更高。

--------------------

> epoll相比于select的改进：
> 
> 1.  **一个进程打开的socket描述符(fd)不受限制，仅受限于操作系统的最大文件句柄数**  
>     select单个线程打开的fd有限，由`FD_SETSIZE`设置，默认1024，可以修改这个宏重新编译内核，但越大，select的效率越低(遍历fd越来越慢)。epoll支持的fd上限是操作系统的最大文件句柄数，受内存影响，可以`cat /proc/sys/fs/file-max`查看。
> 2.  **io效率不会随着fd的数目线性下降**  
>     select/poll会遍历fd。内核实现中epoll根据每个fd的callback函数实现了只对活跃的socket进行操作，从这一点上，epoll实现了一个伪aio。如果所有的socket都处于活跃态，例如告诉lan环境，epoll并不比select/pollx效率高太多，如果过多使用epoll\_ctl，效率还会下降；但是一旦使用wan环境，epoll效率远高于select/poll。
> 3.  **使用mmap加速内核和用户空间的信息传递**  
>     无论是select/poll还是epoll都需要进行内核空间和用户空间的消息传递，如何避免不必要的内存复制就显得非常重要，epoll是通过内核和用户空间mmap同一块内存实现。

nio基础知识

> NIO是NEW IO的简称,不同于传统基于流的io,是一套新的io标准，jdk4出现的nio对文件系统的处理能力不足，jdk7对nio进行了升级，被称nio2.0，提供了aio功能，支持基于文件的异步io和针对网络套接字的异步操作，但是因为nio和aio在操作系统上都是通过epoll实现的，所以实际效率差别不大，netty在提供了几个版本aio的实现后，也不继续支持了。  
> 1、基于块(block),以块为基本单位处理数据  
> 2、为所有原始类型提供buffer支持  
> 3、增加channel对象,作为新的原始io的抽象  
> 4、支持锁和内存映射文件的文件访问接口  
> 5、提供了基于selector的异步网络io，因为jdk使用epoll()代替传统的select()实现，所以没有最大连接句柄的限制，一个Selector可以解除成千上万的客户端

Channel：Channel有四个重要的实现类

*  FileChannel

> 从文件中读写数据

*  DatagramChannel

> 能通过UDP读写网络中的数据

*  SocketChannel

> 能通过TCP读写网络中的数据

*  ServerSocketChannel

> 可以监听新进来的TCP连接,像Web服务器那样。对每一个新进来的连接都会创建一个SocketChannel

*  tips

> jdk4后引入nio的同时也对旧io重写了,旧io类库中的三个类被修改了,用以产生FileChannel。FileInputStream、FileOutputStream、RandomAccessFile。Reader和Writer这种字符模式类不能用来产生通道,字节流和底层nio的性质一样所以可以产生通道,但是channel提供了方法用以在通道中产生Reader、Writer

Buffer

*  Buffer中的三个重要参数

<table> 
 <thead> 
  <tr> 
   <th>参数</th> 
   <th>写模式</th> 
   <th>读模式</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>位置(position)</td> 
   <td>当前缓冲区的位置,将从position的下一个位置写入数据</td> 
   <td>当前缓冲区的位置,将从position的下一个位置读取数据</td> 
  </tr> 
  <tr> 
   <td>容量(capacity)</td> 
   <td>缓存区的总容量上线</td> 
   <td>缓存区的总容量上线</td> 
  </tr> 
  <tr> 
   <td>上限(limit)</td> 
   <td>缓存区的实际上限,总是小于等于容量,通常情况和容量相等</td> 
   <td>代表可读取的总容量,和上次写入的数据量相等</td> 
  </tr> 
 </tbody> 
</table>

Buffer常用方法

*  flip

> 新建buffer时position为0,limit和capacity都是buffer的总容量上限。  
> 读写buffer时,position移动,limit和capacity不变。  
> flipj将buffer从写模式转换为读模式,将limit设为之前position的位置,然后将position重置为0。

public final Buffer flip() {
            limit = position;
            position = 0;
            mark = -1;
            return this;
        }

*  rewind

> position清零,清除mark标志位,用于重新读取buffer

public final Buffer rewind() {
            position = 0;
            mark = -1;
            return this;
        }

*  clear

> position清零,清除mark标志位,将limit设置为capacity的大小,用于再次写入

public final Buffer clear() {
            position = 0;
            limit = capacity;
            mark = -1;
            return this;
        }

*  compact

> compact()方法将所有未读的数据拷贝到Buffer起始处。然后将position设到最后一个未读元素正后面。limit属性依然像clear()方法一样,设置成capacity

### 实际应用 ###

复制文件的三种方式,亲测使用channel的transfer方法效率更高,内部通过内存映射文件实现

FileChannel readChannel = new FileInputStream("压缩文件.zip").getChannel();
    FileChannel writeChannel = new FileOutputStream("压缩文件备份.zip").getChannel();
    ByteBuffer buffer = ByteBuffer.allocate(1024);
    while (readChannel.read(buffer) != -1) {
        buffer.flip();
        writeChannel.write(buffer);
        buffer.clear();
    }//1
    readChannel.transferTo(0, readChannel.size(), writeChannel);//2
    //FileChannel的size()返回关联文件的实际大小
    writeChannel.transferFrom(readChannel, 0, readChannel.size());//3
    writeChannel.close();
    readChannel.close();

通过Selector使用异步io

> 与Selector一起使用时,Channel必须处于非阻塞模式下。这意味着不能将FileChannel与Selector一起使用,因为FileChannel不能切换到非阻塞模式。而套接字通道都可以。实现Channel接口的抽象类SelectableChannel,继承SelectableChannel的Channel才可以使用Selector,因为SelectableChannel才有register()方法

SelectableChannel

`SelectionKey register(Selector sel, int ops, Object att)`

> 第二个参数代表要监听的事件,监听多个事件通过|连接:  
> `SelectionKey.OP_CONNECT`  
> `SelectionKey.OP_ACCEPT`  
> `SelectionKey.OP_READ`  
> `SelectionKey.OP_WRITE(很少注册该事件，该事件仅表示缓冲区是否可用，所以注册后会一直满足条件)`  
> 第三个参数为和Channel绑定的对象,可以将Channel使用的Buffer传入,只传前两个参数也可以,有重载方法

`int validOps()`

> 判断该种Channel支持的监听事件

Selector：

> Selector对象维护了3个SelectionKey的set,一个注册的,一个是就绪的,最后一个是cancel过但是未删除的。最后这个set我们没有方法直接获取到,通过SelectionKey的cancel方法将SelectionKey加入这个set,下次调用select方法就会清空这个set。

`int select()`

> 阻塞到至少有一个通道在你注册的事件上就绪了

`int select(long timeout)`

> 和select()一样,除了最长会阻塞timeout毫秒(参数)

`int selectNow()`

> 不会阻塞,不管什么通道就绪都立刻返回,此方法执行非阻塞的选择操作。如果自从前一次选择操作后,没有通道变成可选择的,则此方法直接返回零。

tips

> select()方法返回的int值表示有多少通道已经就绪。亦即,**自上次调用select()方法后有多少通道变成就绪状态**。如果调用select()方法,因为有一个通道变成就绪状态从而返回了1,但是没有任何处理,再次调用select()方法,如果另一个通道就绪了,它会返回1而不是2  
> 如果第一次调用select()之前就已经有通道就绪了,select()会返回0,但是执行selectedKeys返回的set不为空。select()的返回值是**自上次调用select()方法后有多少通道变成就绪状态**,这一点很重要！

`Selector open()`

> 静态方法,工厂方法,返回Selector实例对象。

`Set<SelectionKey> keys()`

> 返回注册到该Selector上的所有通道的SelectionKey

`Set<SelectionKey> selectedKeys()`

> 一旦调用了select()方法,并且返回值表明有一个或更多个通道就绪了,然后可以通过调用selector的selectedKeys()方法,访问就绪通道

`Selector wakeUp()`

> 某个线程调用select()方法后阻塞了,即使没有通道已经就绪,也有办法让其从select()方法返回。只要让其它线程在调用select()方法的那个Selector对象上执行wakeup()方法即可。阻塞在select()方法上的线程会立马返回。

`close()`

> 用完Selector后调用其close()方法会关闭该Selector,且使注册到该Selector上的所有SelectionKey实例无效,通道本身并不会关闭。

SelectionKey：Channel和Selector之间的关联对象

`int interestOps()`

> 返回代表侦听事件的int,不同的bit代表对应的事件

`int readyOps()`

> 返回代表被侦听的就绪事件的int,不同的bit代表对应的事件

`boolean isAcceptable()`

> 源码为(readyOps() & OP\_ACCEPT) != 0,类似的还有isConnectable()、isReadable()、isWritable()

`SelectableChannel channel()`

> 返回被侦听的Channel

`Selector selector()`

> 返回注册到的Selector

`Object attach(Object ob)`

> 绑定对象,并返回之前的绑定对象

`Object attachment()`

> 返回register()时和Channel一起绑定的或使用attach绑定到Selector的对象

`cancel()`

> 并不直接生效,将该key放到cancelled-key set中,到Selector下次select()时将该key从所有set中删除,但SelectionKey的isValid()会立即回复false

Selector selector = Selector.open();
    ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
    ServerSocket serverSocket = serverSocketChannel.socket( );
    serverSocket.bind (new InetSocketAddress(8888));
    serverSocketChannel.configureBlocking(false);
    serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
    while (true) {
        if (selector.select() == 0) {
            continue;
        }
        for (Iterator<SelectionKey> it = selector.selectedKeys().iterator(); it.hasNext(); it.remove()) {
            SelectionKey key = it.next();
            if (key.isValid() && key.isAcceptable()) {
                //TODO
                ServerSocketChannel server = (ServerSocketChannel) key.channel();
                SocketChannel channel = server.accept( );
                if (channel == null) {
                    continue;
                }
                channel.configureBlocking (false);
                channel.register(selector, SelectionKey.OP_READ, ByteBuffer.allocate(1024));
    
            }
            if (key.isValid() && key.isReadable()) {
                SocketChannel channel = (SocketChannel) key.channel();
                ByteBuffer buffer = (ByteBuffer)key.attachment();
                if (channel.read(buffer) > 0) {
    
                } else {
                    key.cancel();
                }
            }
        }
    }
    selector.close();

总结

1.  SelectionKey维护两个set集合,interestOps和readyOps。Selector维护三个set集合,registeredKeys、selectedKeys、cancelledKeys。每次select()方法调用时,先把cancelledKeys数据同步到registerKeys和selectedKeys,做减法以完成反注册。因为对这几个集合的操作不是线程安全的,所以一般使用Selector的select()只用单线程,而对于select得到的channel和对应的IO操作,可以新开线程或者使用线程池来处理。这也正是IO复用的意义所在。
2.  直接使用jdk的原生nio很麻烦，而且还会碰到bug，多数开发会直接使用netty，性能高，用起来方便，netty5被雪藏了，所以现在还是使用netty4。很多开源工具的通信底层都是netty做的。