java NIO之Selector

Posted 2021-10-09 爱上口袋的天空

1、简介

• 通过一个Selector可以检查更多的通道
• 使用更少的线程来就可以来处理通道了， 相比使用多个线程，避免了线程上下文切换带来的开销

可选择通道： 

• 不是所有的 Channel 都可以被 Selector 复用的。判断他是否继
  承了一个抽象类 SelectableChannel。如果继承了SelectableChannel，则可以被复用，否则不能
• 一个通道可以被注册到多个选择器上，但对每个选择器而言只能被注册一次。通道和选择器之间的关系，使用注册的方式完成。SelectableChannel 可以被注册到Selector 对象上，在注册的时候，需要指定通道的哪些操作，是 Selector 感兴趣的

---

Channel 注册到 Selector：

Channel.register（Selector sel，int ops）一个通道注册到一个选择器时。

第一个参数，指定通道要注册的选择器。
第二个参数指定选择器需要查询的通道操作
供选择器查询的通道

• 可读 : SelectionKey.OP_READ
• 可写 : SelectionKey.OP_WRITE
• 连接 : SelectionKey.OP_CONNECT
• 接收 : SelectionKey.OP_ACCEPT

---

选择键：

（1）Channel 注册到后，并且一旦通道处于某种就绪的状态，就可以被选择器查询到。这个工作，使用选择器 Selector 的 select()方法完成。select 方法的作用，对感兴趣的通道操作，进行就绪状态的查询。
（2）Selector 可以不断的查询 Channel 中发生的操作的就绪状态。并且挑选感兴趣的操作就绪状态。一旦通道有操作的就绪状态达成，并且是 Selector 感兴趣的操作，就会被 Selector 选中，放入选择键集合中

---

1.1、 常用方法

创建一个 Selector 对象：

// 1、获取 Selector 选择器
Selector selector = Selector.open();

注册 Channel 到 Selector：
与 Selector 一起使用时，Channel 必须处于非阻塞模式下

// 1、获取 Selector 选择器
Selector selector = Selector.open();
// 2、获取通道
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
// 3.设置为非阻塞
serverSocketChannel.configureBlocking(false);
// 4、绑定连接
serverSocketChannel.bind(new InetSocketAddress(9999));
// 5、将通道注册到选择器上,并制定监听事件为：“接收”事件
serverSocketChannel.register(selector,SelectionKey.OP_ACCEPT);

---

轮询查询就绪操作：

（1）通过 Selector 的 select（）方法，可以查询出已经就绪的通道操作，这些就绪的状态集合，包存在一个元素是 SelectionKey 对象的 Set 集合中。
（2）下面是 Selector 几个重载的查询 select()方法：

• select():阻塞到至少有一个通道在你注册的事件上就绪了。
• select(long timeout)：和 select()一样，但最长阻塞事件为 timeout 毫秒。
• selectNow():非阻塞，只要有通道就绪就立刻返回。

select()方法返回的 int 值，表示有多少通道已经就绪，更准确的说，是自前一次 select方法以来到这一次 select 方法之间的时间段上，有多少通道变成就绪状态。

例如：首次调用 select()方法，如果有一个通道变成就绪状态，返回了 1，若再次调用select()方法，如果另一个通道就绪了，它会再次返回 1。如果对第一个就绪的channel 没有做任何操作，现在就有两个就绪的通道，但在每次 select()方法调用之间，只有一个通道就绪了。一旦调用 select()方法，并且返回值不为 0 时，在 Selector 中有一个 selectedKeys()方法，用来访问已选择键集合，迭代集合的每一个选择键元素，根据就绪操作的类型，完成对应的操作

Set selectedKeys = selector.selectedKeys();
Iterator keyIterator = selectedKeys.iterator();
while(keyIterator.hasNext()) {
 SelectionKey key = keyIterator.next();
 if(key.isAcceptable()) {
 // a connection was accepted by a ServerSocketChannel.
 } else if (key.isConnectable()) {
 // a connection was established with a remote server.
 } else if (key.isReadable()) {
 // a channel is ready for reading
 } else if (key.isWritable()) {
 // a channel is ready for writing
 }
 keyIterator.remove();
}

---

1.2、实战案例

服务端代码

//服务端代码
@Test
public void serverDemo() throws Exception {
    //1 获取服务端通道
    ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();

    //2 切换非阻塞模式
    serverSocketChannel.configureBlocking(false);

    //3 创建buffer
    ByteBuffer serverByteBuffer = ByteBuffer.allocate(1024);

    //4 绑定端口号
    serverSocketChannel.bind(new InetSocketAddress(8080));

    //5 获取selector选择器
    Selector selector = Selector.open();

    //6 通道注册到选择器，进行监听
    serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

    //7 选择器进行轮询，进行后续操作
    while(selector.select()>0) {
        Set<SelectionKey> selectionKeys = selector.selectedKeys();
        //遍历
        Iterator<SelectionKey> selectionKeyIterator = selectionKeys.iterator();
        while(selectionKeyIterator.hasNext()) {
            //获取就绪操作
            SelectionKey next = selectionKeyIterator.next();
            //判断什么操作
            if(next.isAcceptable()) {
                //获取连接
                SocketChannel accept = serverSocketChannel.accept();

                //切换非阻塞模式
                accept.configureBlocking(false);

                //注册
                accept.register(selector,SelectionKey.OP_READ);

            } else if(next.isReadable()) {
                SocketChannel channel = (SocketChannel) next.channel();

                ByteBuffer byteBuffer = ByteBuffer.allocate(1024);

                //读取数据
                int length = 0;
                while((length = channel.read(byteBuffer))>0) {
                    byteBuffer.flip();
                    System.out.println(new String(byteBuffer.array(),0,length));
                    byteBuffer.clear();
                }

            }

            selectionKeyIterator.remove();
        }
    }
}

客户端代码

//客户端代码
@Test
public void clientDemo() throws Exception {
    //1 获取通道，绑定主机和端口号
    SocketChannel socketChannel =
            SocketChannel.open(new InetSocketAddress("127.0.0.1",8080));

    //2 切换到非阻塞模式
    socketChannel.configureBlocking(false);

    //3 创建buffer
    ByteBuffer byteBuffer = ByteBuffer.allocate(1024);

    //4 写入buffer数据
    byteBuffer.put(new Date().toString().getBytes());

    //5 模式切换
    byteBuffer.flip();

    //6 写入通道
    socketChannel.write(byteBuffer);

    //7 关闭
    byteBuffer.clear();
}

public static void main(String[] args) throws IOException {
    //1 获取通道，绑定主机和端口号
    SocketChannel socketChannel =
            SocketChannel.open(new InetSocketAddress("127.0.0.1",8080));

    //2 切换到非阻塞模式
    socketChannel.configureBlocking(false);

    //3 创建buffer
    ByteBuffer byteBuffer = ByteBuffer.allocate(1024);

    Scanner scanner = new Scanner(System.in);
    while(scanner.hasNext()) {
        String str = scanner.next();

        //4 写入buffer数据
        byteBuffer.put((new Date().toString()+"--->"+str).getBytes());

        //5 模式切换
        byteBuffer.flip();

        //6 写入通道
        socketChannel.write(byteBuffer);

        //7 关闭
        byteBuffer.clear();
    }

}
}

---

1.3、代码参数详解

查看其注册到selector中的源码

  public final SelectionKey register(Selector sel, int ops)
        throws ClosedChannelException
    {
        return register(sel, ops, null);
    }

而且进行轮询的时候本身就是一个集合选择器了
查看其源码也是这样

public abstract Set<SelectionKey> selectedKeys();

遍历该集合的时候使用迭代器进行遍历

---

1.4、步骤总结

1. 创建serverSocketChannel通道，监听端口，并且设置通道为非阻塞模式
2. 创建selector选择器，将其channel注册到选择器上并且监听
3. 调用selector的select方法，进行检测通道的就绪情况
4. 调用selectkeys获取channel集合，便利该集合，判断其事件类型