Java NIO之Selector详细理解

心已赠人 2023-02-17 03:18 15阅读 0赞

### **介绍** ###

Selector一般称为选择器。它是Java NIO核心组件之一，选择器管理着一个被注册的通道集合的信息和它们的就绪状态。通道是和选择器一起被注册的，并且使用选择器来更新通道的就绪状态。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NraDIwMTVqYXZh_size_16_color_FFFFFF_t_70][]

**可选择通道(SelectableChannel)**

抽象类SelectableChannel提供了实现通道的可选择性所需要的公共方法。继承该SelectableChannel的通道(如：socket通道)都是可选择的，因此可以配置成非阻塞模式。

//可以注册的Channel的抽象
    
    public abstract class SelectableChannel
    
        extends AbstractInterruptibleChannel
    
        implements Channel
    
    {
        //返回创建此通道的Provider
        public abstract SelectorProvider provider();
    
    
        //返回注册的SelectionKey（即事件的操作类型）
        public abstract int validOps();
    
    
        //判断当前Channel是否有注册到任意的Selector
        public abstract boolean isRegistered();
    
    
        //获取该Channel注册在Selector的SelectionKey(事件类型)
        public abstract SelectionKey keyFor(Selector sel);
    
    
         //将Channel注册到给定的Selector sel上，并指定注册的事件类型(SelectionKey.OP_READ,OP_WRITE,OP_CONNECT,OP_ACCEPT等)
    
        //attachment 为附属的resulting key，可以为null
        //只有非阻塞模型的Channel才能调用register方法
        public abstract SelectionKey register(Selector sel, int ops, Object att)
    
    
        //将Channel注册到给定的Selector sel上，并指定注册的事件类型(SelectionKey.OP_READ,OP_WRITE,OP_CONNECT,OP_ACCEPT等)
        public final SelectionKey register(Selector sel, int ops)
    
    
        //配置是否是阻塞模式
        public abstract SelectableChannel configureBlocking(boolean block)
    
    
        //获取当前是否是阻塞模式
        public abstract boolean isBlocking();
    
    
        //获取 configureBlocking和register方法同步的锁
        public abstract Object blockingLock();
    
    }

**选择键(SelectionKey)**

选择键封装了特定的通道与特定的选择器的注册关系。选择键对象被SelectableChannel.register()返回并提供一个表示这种注册关系的标记。

SelectionKey常用的API如下

//选择键封装了特定的通道与特定的选择器的注册关系
    
    public abstract class SelectionKey {
    
               //返回创建该Key的Channel对象，即使当前key被取消也会返回
               public abstract SelectableChannel channel();
      
    
               //返回创建该key的Selector
                public abstract Selector selector();
             
    
                 //判断该key是否可用
                 public abstract boolean isValid();
    
    
                  //请求取消此键的通道向其选择器的注册。返回时，该密钥将无效，并将被添加到其选择器的已取消密钥集中。在下一次选择操作期间，该键将从所有选择器的键集中删除。
               public abstract void cancel();
    
    
               //  获取此键的 interest 集合。
               public abstract int interestOps();
    
    
               //将此键的 interest 集合设置为给定值。
               public abstract SelectionKey interestOps(int ops);
    
    
               //获取该通道已经就绪的操作
               public abstract int readyOps();
    
    
               public static final int OP_READ = 1 << 0;  //读操作事件
               public static final int OP_WRITE = 1 << 2;  //写操作事件
               public static final int OP_CONNECT = 1 << 3;  //连接事件
               public static final int OP_ACCEPT = 1 << 4;  //接受事件
    
    
               //测试此键的通道是否已准备好进行读取。
                public final boolean isReadable()
    
    
               //测试此键的通道是否已准备好进行写入。
               public final boolean isWritable()
    
    
               // 测试此键的通道是否已完成其套接字连接操作。
               public final boolean isConnectable()
    
    
                // 测试此键的通道是否已准备好接受新的套接字连接。
                 public final boolean isAcceptable()
    
    
               //将附加对象绑定到SelectionKey上，便于识别给定的通道
               public final Object attach(Object ob)
    
    
                //取出绑定在SelectionKey上的附加对象
                 public final Object attachment()
    }

jdk1.8版本SelectionKey有4中事件，OP\_READ(读事件)、OP\_WRITE(写事件)、OP\_CONNECT(连接事件)和OP\_ACCEPT(接受事件)。

**Selector选择器**

Selector常用的API

public abstract class Selector implements Closeable {
    
        //创建一个选择器
        public static Selector open()
    
    
        //获取当前选择器是否已打开
        public abstract boolean isOpen()
    
    
        //返回创建此通道的提供者
        public abstract SelectorProvider provider()
    
    
       //返回此选择器的键集
        public abstract Set<SelectionKey> keys()
    
    
       //返回此选择器已选择的键
        public abstract Set<SelectionKey> selectedKeys()
    
    
       //获取已经I/O准备就绪的键集，不会阻塞
        public abstract int selectNow()
    
    
        //阻塞直到注册到Selector上的Channel有事件发生，或者到了超时时间
        //指定最长等待阻塞的时间，单位ms
    
        public abstract int select(long timeout)
    
    
    
        //阻塞直到注册在Selector中的Channel 发送可读写事件(或其他注册事件)
        public abstract int select()
    
        //唤醒调用select()阻塞的线程
        public abstract Selector wakeup()
    
    
        //关闭Selector
        public abstract void close()
    
    }

**open方法**

open()方法用于创建一个Selector对象

Selector selector = Selector.open();

**将Channel注册到Selector中**

我们需要将 Channel 注册到Selector 中，这样才能通过 Selector 监控 Channel 中的事件，注：一个Channel要注册到Selector中，需要先将这个Channel设置成非阻塞模式。

//设置成非阻塞模式

channel.configureBlocking(false);

SelectionKey key = channel.register(selector, SelectionKey.OP\_READ);

**select****方法**

select()和select(long timeout)，select方法是Selector的核心方法，这两个方法会阻塞线程。

select(long timeout)：阻塞直到注册到Selector上的Channel有事件发生，或者到了超时时间指定最长等待阻塞的时间，单位ms

select()：阻塞直到注册在Selector中的Channel 发送可读写事件(或其他注册事件)

除了正常收到事件或超时还有三种方法可以唤醒在select()中阻塞的线程

1. wakeup()

2. close()

3. interrupt()

**获取就绪的Channel**

Set<SelectionKey> selectedKeys = selector.selectedKeys();
    Iterator<SelectionKey> keyIterator = selectedKeys.iterator();
    
    
    while(keyIterator.hasNext()) {
        SelectionKey key = keyIterator.next();
        keyIterator.remove();
    
    
        //可能有多个注册事件就绪
        if(key.isAcceptable()) {
            // a connection was accepted by a ServerSocketChannel.
        }
    
    
        if (key.isConnectable()) {
            // a connection was established with a remote server.
        }
    
    
    
        if (key.isReadable()) {
            // a channel is ready for reading
        }
    
    
        if (key.isWritable()) {
            // a channel is ready for writing
        }
    
    }

注意, 在每次迭代时, 我们都调用 “keyIterator.remove()” 将这个 key 从迭代器中删除, 因为 select() 方法仅仅是简单地将就绪的 IO 操作放到 selectedKeys 集合中, 因此如果我们从 selectedKeys 获取到一个 key, 但是没有将它删除, 那么下一次 select 时, 这个 key 所对应的 IO 事件还在 selectedKeys 中.

**attach 和 attachment 方法**

public final Object attach(Object ob)  将附加对象绑定到SelectionKey上，便于识别给定的通道

public final Object attachment()  取出绑定在SelectionKey上的附加对象

ServerSocketChannel和Selector使用示例：

import java.io.IOException;
    import java.net.InetSocketAddress;
    import java.net.ServerSocket;
    import java.nio.ByteBuffer;
    import java.nio.channels.SelectionKey;
    import java.nio.channels.Selector;
    import java.nio.channels.ServerSocketChannel;
    import java.nio.channels.SocketChannel;
    import java.nio.charset.Charset;
    import java.util.Iterator;
    import java.util.LinkedList;
    import java.util.List;
    
    
    
    public class SelectSocketsThreadPool {
        private static final int MAX_THREADS = 5;  //线程池数量
        public static int PORT_NUMBER = 7786;  //指定默认端口号
        private ThreadPool  pool = new ThreadPool(MAX_THREADS);
    
    
    
        public static void main(String[] args) throws Exception {
                new SelectSocketsThreadPool().go(args);
        }
    
    
    
        public void go(String [] args) throws Exception {
            int port = PORT_NUMBER;
            if (args.length > 0){
                port = Integer.parseInt(args[0]);
            }
            System.out.println("Listening on port "+port);
    
    
    
            //初始化ServerSocketChannel
            ServerSocketChannel serverChannel = ServerSocketChannel.open();
    
            //该ServerSocketChannel对应的ServerSocket
            ServerSocket serverSocket = serverChannel.socket();
    
            //初始化一个Selector
            Selector selector = Selector.open();
    
    
            serverSocket.bind(new InetSocketAddress(port));  //绑定端口
            serverChannel.configureBlocking(false);  //非阻塞
    
    
    
            //绑定 ACCEPT事件
            serverChannel.register(selector, SelectionKey.OP_ACCEPT);
    
    
            while (true){
                //阻塞直到有新的连接
                int n= selector.select();
                if (n == 0){
                    continue;
                }
    
                Iterator it = selector.selectedKeys().iterator();
    
                while (it.hasNext()){
                    SelectionKey key = (SelectionKey)it.next();
                    if (key.isAcceptable()) { //key是ACCEPT
                        System.out.println("accept connection!");
    
                        //获取该key对应的Channel
                        System.out.println();
                        ServerSocketChannel server = (ServerSocketChannel)key.channel();
    
                        //接收到的SocketChannel
                        SocketChannel channel = server.accept();
                        registerChannel(selector,channel,SelectionKey.OP_READ);
                        sayHello(channel);
                    }
    
    
    
                    if (key.isReadable()){
                        readDataFromSocket(key);
                    }
    
    
                    it.remove();
                }
            }
        }
    
    
    
        private void sayHello(SocketChannel channel) throws IOException {
            buffer.clear();
            buffer.put("Hi there!\r\n".getBytes());
            buffer.flip();
            channel.write(buffer);
        }
    
    
    
    
    
        private ByteBuffer buffer = ByteBuffer.allocateDirect(1024);
    
    
    
        //连接上来的SocketChannel，注册到当前Selector上，注册的操作为READ
        protected void registerChannel(Selector selector, SocketChannel channel, int opRead) throws IOException {
            if (channel == null){
               return;
            }
    
    
            channel.configureBlocking(false);
            channel.register(selector,opRead);
        }
    
    
    
    
    
        //读取数据
        protected void readDataFromSocket(SelectionKey  key) {
            WorkThread workThread = pool.getWorker();
            if (workThread == null){
                return;
            }
    
    
            workThread.serviceChannel(key);
        }
    
    
    
    
    
    
    
        class ThreadPool {
            List idle = new LinkedList();
            //创建指定数量的线程并添加到空闲队列idle中
            ThreadPool(int poolSize) {
                for (int i = 0 ; i < poolSize; i++){
                    WorkThread  thread = new WorkThread(this);
                    thread.setName("Worker" + (i+1));
                    thread.start();
                    idle.add(thread);
               }
            }
    
    
    
    
    
            //从队列中获取空闲线程
            WorkThread getWorker(){
                WorkThread workThread = null;
                synchronized (idle){
                    if (idle.size() > 0){
                        workThread = (WorkThread)idle.remove(0);
                    }
                }
                return workThread;
            }
    
    
    
            //将线程返还到空闲线程
            void returnWorker(WorkThread workThread){
                synchronized (idle){
                    idle.add(workThread);
                }
            }
        }
    
    
    
    
    
        //线程执行的任务类
    
        class WorkThread extends Thread {
            private ByteBuffer  buffer = ByteBuffer.allocate(1024);
            private ThreadPool pool;
            private SelectionKey key;
    
    
            WorkThread(ThreadPool pool){
                this.pool = pool;
            }
    
    
            @Override
            public synchronized void run() {
                System.out.println(this.getName() +" is ready");
                while (true){
                    try{
                       this.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
    
    
                    if (key == null){
                       continue;
                    }
    
    
                    System.out.println(this.getName() +" has been awakened");
                    try{
                        drainChannel(key);  //处理就绪的key
                    }catch (Exception e){
                        System.out.println(" caught "+ e + "closing channel");
    
                        try {
                            key.channel().close();
                        } catch (IOException e1) {
                            e1.printStackTrace();
                        }
                    }
    
    
                    key = null;
                    this.pool.returnWorker(this);  //线程处理完任务放回空闲队列
                }
            }
    
    
    
            synchronized void serviceChannel(SelectionKey key){
                this.key = key;
                key.interestOps(key.interestOps() & (~SelectionKey.OP_READ));
                this.notify();
            }
    
    
    
            //处理就绪的key
            private void drainChannel(SelectionKey key) throws IOException {
                SocketChannel channel = (SocketChannel)key.channel();
                int count;
                buffer.clear();
                while ((count = channel.read(buffer)) > 0 ){
                    System.out.println(new String(buffer.array(),0,count));
                    buffer.flip();
                    while (buffer.hasRemaining()){
                        channel.write(buffer);
                    }
    
                    buffer.clear();
                }
    
    
                if (count < 0){
                    channel.close();
                    return;
                }
    
    
                key.interestOps(key.interestOps() | SelectionKey.OP_READ);
                key.selector().wakeup();
            }
    
    
    
            private String decode(ByteBuffer bb) {
                Charset charset = Charset.forName("ASCII");
                return charset.decode(bb).toString();
            }
    
        }
    
    }

感谢：

《Java NIO》

[https://blog.csdn.net/u014634338/article/details/82865622][https_blog.csdn.net_u014634338_article_details_82865622]

[http://tutorials.jenkov.com/java-nio/selectors.html][http_tutorials.jenkov.com_java-nio_selectors.html]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NraDIwMTVqYXZh_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200611185138521.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NraDIwMTVqYXZh,size_16,color_FFFFFF,t_70
[https_blog.csdn.net_u014634338_article_details_82865622]: https://blog.csdn.net/u014634338/article/details/82865622
[http_tutorials.jenkov.com_java-nio_selectors.html]: http://tutorials.jenkov.com/java-nio/selectors.html