JAVA NIO的基本类

我会带着你远行 2022-04-23 05:00 148阅读 0赞

> Netty虽然对JAVA NIO 进行了封装，要了解Netty的源码，还是有必要对JAVA的NIO进行了解。以下的类是比较重要的

# Selector #

多路复用器，可以理解为多个channel注册到一个selector,selector可以轮询这些channel。一个channel注册到selector会返回一个注册凭证即(`SelectionKey`)。一个多路复用器会维护3种set如下

1.  keys() 返回所有注册凭证(`SelectionKey`)
2.  selectedKeys() 返回的所有有感兴趣事件的注册凭证(`SelectionKey`)
3.  cancelled-key 删除的注册凭证(`SelectionKey`)

##### 多路复用器的select过程 #####

触发一次多路复用器的select过程，有以下3种方法

*  select() 如果没有可操作的事件阻塞等待
 *  select(long) 如果没有可操作的事件阻塞等待多久
 *  selectNow() 不会阻塞，立即返回

一次select过程会分为以下3步

1.  selector的cancelled-key set 里的`SelectionKey`会被移除，也就是对应的channel从selector里移除。
2.  当selector轮询到的channel里的事件是它的`SelectionKey`里感兴趣的事件，这个`SelectionKey`就会加入到selector的selectedKeys set里.这个过程又分成2种情况

*  情况1：如果channel对应的`SelectionKey`没有加入到selector的selectedKeys set里。那`SelectionKey`里的ready-operation set会被更新成对应的channel里的事件，而`SelectionKey`里的ready-operation set之前的值会被丢弃
 *  情况2: 如果channel对应的`SelectionKey`已经加入到selector的selectedKeys set里。SelectionKey里的ready-operation set会合并当前的channel里的事件，`SelectionKey`里的ready-operation set之前的值不会被丢弃

1.  如果在2的处理过程里，`SelectionKey`被删除了，那会执行1

当一个线程在执行selector.select()或selector.select(long)方法阻塞等待时.可以通过另外线程调用以下三种方法来唤醒

*  selector.wakeup()
 *  selector.close()
 *  调用阻塞线程的interrupt()

> 注意Selector里key 和 selected-key sets不是线程安全的，在多线程的情况下需要考虑到同步set. 例如当selected-key sets同时被2个线程获取到，其中第一个线程在iterator的过程中，另个线程修改了sets里的东西(除了调用remove操作)，这样第一个线程会立即抛异常。

# SelectableChannel #

SelectableChannel可以注册到selector上，当注册完成后会返回一个注册凭证(`SelectionKey`).  
新建的`SelectableChannel`是阻塞的，可以通过设置configureBlocking(false)改成非阻塞。

> 如果需要注册到多路复用器配合，是需要改成非阻塞模式

**方法**

*  configureBlocking() 配置阻塞或非阻塞
 *  isBlocking() 检查当前的channel是哪种模式

**ServerSocketChannel的子类**

*  ServerSocketChannel 服务器channel
 *  SocketChannel 客户端channel  
    这2个也是 abstract,实现类分别是  
    `ServerSocketChannelImpl`和`SocketChannel`

ServerSocketChannel通过  
`SelectorProvider.provider().openSocketChannel();`  
获取  
SocketChannel通过  
`SelectorProvider.provider().openServerSocketChannel();`  
获取

# SelectionKey #

SelectableChannel注册到selector 后,返回的对象是`SelectionKey`,代表注册的信息，理解为一个注册的凭证（token）.  
在channel.register(Selector sel, int ops, Object att)方法里，其中的int ops代表的是感兴趣的操作，可以后续通过interestOps(int ops)来改变

`SelectionKey`里含有2个key set,

1.  interest set即interestOps()返回值,这里是指在selector执行select()的时候,告诉selector感兴趣的事件。可以通过interestOps(int)来改变
2.  ready set即readyOps()返回值，是interest set里的子集。

以上返回的都是一个int类型，int类型的每一位都代表了事件类型，事件类型包括了

*  OP\_READ = 1 << 0
 *  OP\_WRITE = 1 << 2
 *  OP\_CONNECT = 1 << 3
 *  OP\_ACCEPT = 1 << 4

上面的几个事件结合TCP的连接处理图就会清晰的多  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2d1b194bA_size_16_color_FFFFFF_t_70]

**方法**

*  cancel() 使得凭证失效，不会立即失效，而是放到selector的删除key集合（cancelled-key set）中等待下轮selector循环再删除这个注册channel
 *  interestOps() 返回感兴趣的事件
 *  interestOps(int) 注册感兴趣的事件
 *  readyOps() 返回感兴趣的事件里的已经就绪的事件

# 例子 #

> 该例子来源于https://segmentfault.com/a/1190000006824196  
> 结合上面的知识点现在就很好理解下面的代码了

public class NioEchoServer {
        private static final int BUF_SIZE = 256;
        private static final int TIMEOUT = 3000;
    
        public static void main(String args[]) throws Exception {
            // 打开服务端 Socket
            ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
    
            // 打开 Selector
            Selector selector = Selector.open();
    
            // 服务端 Socket 监听8080端口, 并配置为非阻塞模式
            serverSocketChannel.socket().bind(new InetSocketAddress(8080));
            serverSocketChannel.configureBlocking(false);
    
            // 将 channel 注册到 selector 中.
            // 通常我们都是先注册一个 OP_ACCEPT 事件, 然后在 OP_ACCEPT 到来时, 再将这个 Channel 的 OP_READ
            // 注册到 Selector 中.
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
    
            while (true) {
                // 通过调用 select 方法, 阻塞地等待 channel I/O 可操作
                if (selector.select(TIMEOUT) == 0) {
                    System.out.print(".");
                    continue;
                }
    
                // 获取 I/O 操作就绪的 SelectionKey, 通过 SelectionKey 可以知道哪些 Channel 的哪类 I/O 操作已经就绪.
                Iterator<SelectionKey> keyIterator = selector.selectedKeys().iterator();
    
                while (keyIterator.hasNext()) {
    
                    SelectionKey key = keyIterator.next();
    
                    // 当获取一个 SelectionKey 后, 就要将它删除, 表示我们已经对这个 IO 事件进行了处理.
                    keyIterator.remove();
    
                    if (key.isAcceptable()) {
                        // 当 OP_ACCEPT 事件到来时, 我们就有从 ServerSocketChannel 中获取一个 SocketChannel,
                        // 代表客户端的连接
                        // 注意, 在 OP_ACCEPT 事件中, 从 key.channel() 返回的 Channel 是 ServerSocketChannel.
                        // 而在 OP_WRITE 和 OP_READ 中, 从 key.channel() 返回的是 SocketChannel.
                        SocketChannel clientChannel = ((ServerSocketChannel) key.channel()).accept();
                        clientChannel.configureBlocking(false);
                        //在 OP_ACCEPT 到来时, 再将这个 Channel 的 OP_READ 注册到 Selector 中.
                        // 注意, 这里我们如果没有设置 OP_READ 的话, 即 interest set 仍然是 OP_CONNECT 的话, 那么 select 方法会一直直接返回.
                        clientChannel.register(key.selector(), OP_READ, ByteBuffer.allocate(BUF_SIZE));
                    }
    
                    if (key.isReadable()) {
                        SocketChannel clientChannel = (SocketChannel) key.channel();
                        ByteBuffer buf = (ByteBuffer) key.attachment();
                        long bytesRead = clientChannel.read(buf);
                        if (bytesRead == -1) {
                            clientChannel.close();
                        } else if (bytesRead > 0) {
                            key.interestOps(OP_READ | SelectionKey.OP_WRITE);
                            System.out.println("Get data length: " + bytesRead);
                        }
                    }
    
                    if (key.isValid() && key.isWritable()) {
                        ByteBuffer buf = (ByteBuffer) key.attachment();
                        buf.flip();
                        SocketChannel clientChannel = (SocketChannel) key.channel();
    
                        clientChannel.write(buf);
    
                        if (!buf.hasRemaining()) {
                            key.interestOps(OP_READ);
                        }
                        buf.compact();
                    }
                }
            }
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2d1b194bA_size_16_color_FFFFFF_t_70]: /images/20220323/2f75fd732f38497fa67806571a7c4d82.png