【NIO】Selector：API 及使用实例

深碍√TFBOYSˉ_ 2022-12-30 03:44 108阅读 0赞

Selector（选择器）是Java NIO中能够检测一到多个NIO通道，并能够知晓通道是否为诸如读写事件做好准备的组件。这样，一个单独的线程可以管理多个channel，从而管理多个网络连接。NIO 模型的 Selector 就像一个大总管，负责监听各种IO事件，然后转交给后端线程去处理。  
![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center][]

从图中可以看出，当有读或写等任何注册的事件发生时，可以从Selector 中获得相应的 **SelectionKey**，同时从 SelectionKey 中可以找到发生的事件和该事件所发生的具体的 SelectableChannel，以获得客户端发送过来的数据。

==> 再来对比一下 BIO 和 NIO：

对于 BIO ，我们最直观的感受就是各种 Stream，后端线程需要阻塞等待客户端写数据(比如read方法)，如果客户端不写数据线程就要阻塞。

对于 NIO，我们最直观的感受就是设置为 noblocking 的 Channel，它将等待客户端操作的事情交给了大总管 Selector，Selector 负责轮询所有已注册的客户端，只有当客户端主动触发了现有的事件时，Selector 才会让后端线程去读写 Channel。所以后端线程不需要做任何阻塞等待，直接处理客户端事件的数据即可；处理完马上结束，或返回线程池供其他客户端事件继续使用。

### Selector 基本 API ###

Selector 的常用方法见下表：

<table> 
 <thead> 
  <tr> 
   <th align="left">方 法</th> 
   <th align="left">描 述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">Set keys()</td> 
   <td align="left">所有的 SelectionKey 集合。代表注册在该Selector上的Channel</td> 
  </tr> 
  <tr> 
   <td align="left">selectedKeys()</td> 
   <td align="left">被选择的 SelectionKey 集合。返回此Selector的已选择键集</td> 
  </tr> 
  <tr> 
   <td align="left">int select()</td> 
   <td align="left">监控所有注册的Channel，当它们中间有需要处理的 IO 操作时，该方法返回，并将对应得的 SelectionKey 加入被选择的SelectionKey集合中，该方法返回这些 Channel 的数量。</td> 
  </tr> 
  <tr> 
   <td align="left">int select(long timeout)</td> 
   <td align="left">可以设置超时时长的 select() 操作</td> 
  </tr> 
  <tr> 
   <td align="left">int selectNow()</td> 
   <td align="left">执行一个立即返回的 select() 操作，该方法不会阻塞线程</td> 
  </tr> 
  <tr> 
   <td align="left">Selector wakeup()</td> 
   <td align="left">使一个还未返回的 select() 方法立即返回</td> 
  </tr> 
  <tr> 
   <td align="left">void close()</td> 
   <td align="left">关闭该选择器</td> 
  </tr> 
 </tbody> 
</table>

可以看到上面还提到了一个很重要的概念 --SelectionKey，它里面定义了四种事件：

public static final int OP_READ = 1 << 0;    // 1，读
    public static final int OP_WRITE = 1 << 2;   // 4，写
    public static final int OP_CONNECT = 1 << 3; // 8，连接（Client）
    public static final int OP_ACCEPT = 1 << 4;  // 16，接收（Server）

SelectionKey 常用方法如下：

<table> 
 <thead> 
  <tr> 
   <th align="left">方 法</th> 
   <th align="left">描 述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">int interestOps()</td> 
   <td align="left">获取感兴趣事件集合</td> 
  </tr> 
  <tr> 
   <td align="left">int readyOps()</td> 
   <td align="left">获取通道已经准备就绪的操作的集合</td> 
  </tr> 
  <tr> 
   <td align="left">SelectableChannel channel()</td> 
   <td align="left">获取注册通道</td> 
  </tr> 
  <tr> 
   <td align="left">Selector selector()</td> 
   <td align="left">返回选择器</td> 
  </tr> 
  <tr> 
   <td align="left">boolean isReadable()</td> 
   <td align="left">检测 Channal 中读事件是否就绪</td> 
  </tr> 
  <tr> 
   <td align="left">boolean isWritable()</td> 
   <td align="left">检测 Channal 中写事件是否就绪</td> 
  </tr> 
  <tr> 
   <td align="left">boolean isConnectable()</td> 
   <td align="left">检测 Channel 中连接是否就绪</td> 
  </tr> 
  <tr> 
   <td align="left">boolean isAcceptable()</td> 
   <td align="left">检测 Channel 中接收是否就绪</td> 
  </tr> 
 </tbody> 
</table>

### NIO服务器示例 ###

使用 NIO 中非阻塞I/O编写服务器处理程序，大体上可以分为下面三个步骤：

1）向 Selector 对象注册感兴趣的事件

private Selector getSelector() throws Exception {
        
        // 创建Selector对象
        Selector sel = Selector.open();
    	
        // 创建可选择通道，并设置为非阻塞模式
        // 注：1.这个server自己的通道，用来接收连接（Accept）的
        //     2.处理其余事件的Channel可以根据连接获取（accept），然后再注册相应事件
        ServerSocketChannel server = ServerSocketChannel.open();
        // 必须配置为非阻塞，该Channel才能往selector上注册，否则会报错，selector模式本身就是非阻塞模式 
        server.configureBlocking(false);
    	
        // 绑定通道到指定端口
        ServerSocket socket = server.socket();
        socket.bind(new InetSocketAddress(8080));
    	
        // 向Selector中注册感兴趣的事件（这里的ACCEPT就是新连接发生时所产生的事件）
        // 注：对于ServerSocketChannel 通道来说，我们唯一可以指定的参数就是OP_ACCEPT
        server.register(sel, SelectionKey.OP_ACCEPT);
        return sel;
    }

2）从Selector 中获取感兴趣的事件，即开始监听，进入内部循环

public void listen() throws IOException {
        
        System.out.println("listen on" + port);
        // while（true）持续接受连接
        while (true) {
        
        	// 该调用会阻塞，直至至少有一个事件发生
            // 1.当有客户端来连接时，就会触发 ServerSocketChannel 的 ACCEPT 事件
            // 2.当客户端发送来消息时，就会触发 ServerSocketChannel 的 READ 事件
            // 3.当客户端读取了发送的消息时，就会触发 ServerSocketChannel 的 WRITE 事件
            selector.select();
            // 获取发生事件的SelectionKey
            Set<SelectionKey> keys = selector.selectedKeys();
            // 再使用迭代器进行循环已发生的事件
            Iterator<SelectionKey> iter = keys.iterator();
            while (iter.hasNext()) {
        
                SelectionKey key = iter.next();
                // 根据事件类型调用相应函数进行处理
                process(key);
                // 移除已经处理过的事件，避免重复
                iter.remove();
            }
        }
    }

3）根据不同的事件进行相应的处理

private void process(SelectionKey key) throws IOException {
        
        // 接收请求
        if (key.isAcceptable()) {
        
        	// 通过key拿到注册当前事件的Channel（因为是ACCEPT，所以只能是ServerSocketChannel ）
            ServerSocketChannel server = (ServerSocketChannel) key.channel();
            // accept返回一个包含新连接的SocketChannel，将会为当前client提供服务
            // 注：这个方法不会阻塞，因为SeverSocketChannel设置了非阻塞，直接返回null
            SocketChannel channel = server.accept();
            channel.configureBlocking(false);
            // 将该socketChannel注册到Selector，绑定上READ事件
            channel.register(selector, SelectionKey.OP_READ);
        }
        // 读事件
        else if (key.isReadable()) {
        
        	// 通过key拿到当前连接的 SocketChannel
            SocketChannel channel = (SocketChannel) key.channel();
            // 将通道中的数据读到缓冲区 buffer，返回读了多少字节
            int len = channel.read((ByteBuffer) buffer);
            if (len > 0) {
        
    	     	// 通过 buffer.array() 拿到buffer中的数组（没必要buffer.filp）
                String content = new String(((ByteBuffer) buffer).array(), 0, len);
                // 将当前SocketChannel再注册上WRITE事件
                SelectionKey skey = channel.register(selector, SelectionKey.OP_WRITE);
                skey.attach(content);
            } else {
        
                channel.close();
            }
            buffer.clear();
        }
    	// 写事件
        else if (key.isWritable()) {
        
       		// 拿到当前连接的 SocketChannel
            SocketChannel channel = (SocketChannel) key.channel();
            // 将要发送给客户端的内容先写到缓冲区
            String content = (String) key.attachment();
            writeBuffer.put(("输出内容：" + content).getBytes());
            // 这里必须 filp()，因为 put() 后 position 等于消息的length，所以client拿到后无法再从这个缓冲区读出数据
            sendBuffer.flip();
            
            if(block != null){
         
            	// 将缓冲区的数据写到SocketChannel
                channel.write(block); 
            }else{
         
                channel.close();
            }
        }
    }

此处分别判断是接受请求、读数据还是写事件，分别作不同的处理。

> PS：如果当前管道 Channel 对不止一种事件感兴趣，那么可以用“位或”操作符将常量连接起来 `channel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE)`;

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/3ef59677037d4c548674af73e038e605.png