Selector类的使用(一)

水深无声 2021-08-30 01:01 306阅读 0赞

### Selector类的使用 ###

Selector类的主要作用是作为SelectableChannel对象的多路复用器。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70]

可通过调用Selector类的`open()`方法创建选择器，该方法将使用系统的默认SelectorProvider创建新的选择器。也可通过调用自定义选择器提供者的`openSelector()`方法来创建选择器。在通过选择器的`close()`方法关闭选择器之前，选择器一直保持打开状态。

通过SelectionKey对象来表示SelectableChannel (可选择通道)到选择器的注册。选择器维护了3种SelectionKey-Set (选择键集)。

1.  `键集`:包含的键表示当前通道到此选择器的注册，也就是通过某个通道的register()方法注册该通道时，所带来的影响是向选择器的键集中添加了一个键。此集合由keys()方法返回。键集本身是不可直接修改的。
2.  `已选择键集`:在首先调用select()方法选择操作期间，检测每个键的通道是否已经至少为该键的相关操作集所标识的一个操作准备就绪，然后调用selectedKeys() 方法返回已就绪键的集合。已选择键集始终是键集的一个子集。
3.  `已取消键集`:表示已被取消但其通道尚未注销的键的集合。不可直接访问此集合。已取消键集始终是键集的一个子集。在select()方法选择操作期间，从键集中移除已取消的键。

在新创建的选择器中，这3个集合都是空集合。

无论是通过关闭某个键的通道还是调用该键的cancel)方法来取消键，该键都被添加到其选择器的已取消键集中。取消某个键会导致`在下一次select()方法选择操作期间注销该键的通道，而在注销时将从所有选择器的键集中移除该键。`

通过select()方法选择操作将键添加到已选择键集中。可通过调用已选择键集的remove()方法，或者通过调用从该键集获得的iterator的remove()方法直接移除某个键。通过任何其他方式都无法直接将键从已选择键集中移除，特别是，它们不会因为影响选择操作而被移除。不能将键直接添加到已选择键集中。

select()、select(long)和selectNow()方法：

在每次select()方法选择操作期间，都可以将键添加到选择器的已选择键集或从中将其移除，并且可以从其键集和已取消键集中将其移除。选择是由select()、select(long) 和selectNow()方法执行的，涉及以下3个步骤。

1.  将已取消键集中的每个键从所有键集中移除(如果该键是键集的成员),并注销其通道。此步骤使已取消键集成为空集。
2.  在开始进行select()方法选择操作时，应查询基础操作系统来更新每个剩余通道的准备就绪信息，以执行由其键的相关集合所标识的任意操作。对于已为至少一个这样的操作准备就绪的通道，执行以下两种操作之一。
    
    1.  如果该通道的键尚未在已选择键集中，则将其添加到该集合中，并修改其准备就绪操作集，以准确地标识那些通道现在已报告为之准备就绪的操作。丢弃准备就绪操作集中以前记录的所有准备就绪信息。
    2.  如果该通道的键已经在已选择键集中，则修改其准备就绪操作集，以准确地标识所有通道已报告为之准备就绪的新操作。保留准备就绪操作集以前记录的所有准备就绪信息。换句话说，基础系统所返回的准备就绪集是和该键当前准备就绪操作集按位分开（bitwise-disjoined）的。  
        如果在此步骤开始时键集中的所有键都为空的相关集合，则不会更新已选择键集合任意键的准备就绪操作集。
    3.  如果在步骤2进行时已将任何键添加到已取消的键集，则将它们按照步骤1进行处理。

`是否阻塞选择操作以等待一个或多个通道准备就绪，以及要等待多久，是这3种选择方法之间的本质差别。`

并发：

`选择器自身可由多个并发线程安全使用，但是其键集并非如此。`

选择操作在选择器本身上、在键集上和在已选择键集上是同步的，顺序也与此顺序相同。在执行，上面的`步骤1 )和步骤3)时，它们在已取消键集上也是同步的`。

`在执行选择操作的过程中，更改选择器键的相关集合对该操作没有影响;进行下一次选择操作才会看到此更改。`

可在任意时间取消键和关闭通道。因此，在一个或多个选择器的键集中出现某个键并不意味着该键是有效的，也不意味着其通道处于打开状态。如果存在另一个线程取消某个键或关闭某个通道的可能性，那么应用程序代码进行同步时应该小心，并且必要时应该检查这些条件。

阻塞在select()或select(long)方法中的某个线程可能被其他线程以下列3种方式之一中断:

1.  通过调用选择器的wakeup()方法;
2.  通过调用选择器的close()方法;
3.  在通过调用已阻塞线程的interrupt()方法的情况下，将设置其中断状态并且将调用该选择器的wakeup()方法。

close()方法在选择器上是同步的，并且所有3个键集都与选择操作中的顺序相同。

一般情况下，选择器的键和已选择键集由多个并发线程使用是不安全的。如果这样的线程可以直接修改这些键集之一那么应该通过对该键集本身进行同步来控制访问。

`这些键集的iterator()方法所返回的迭代器是快速失败的:如果在创建迭代器后以任何方式(调用迭代器自身的remove()方法除外)修改键集，则会抛出ConcurrentModificationException。`

#### 验证select()方法具有阻塞性 ####

`public abstract int select()` 方法的作用是选择一组键， 其相应的通道已为I/O操作准备就绪。此方法执行处于阻塞模式的选择操作。仅在至少选择一个通道、调用此选择器的wakeup()方法，或者当前的线程已中断(以先到者为准)后，此方法才返回。`返回值代表添加到就绪操作集的键的数目，该数目可能为零，为零代表就绪操作集中的内容并没有添加新的键，保持内容不变。`

public static void main(String[] args) { 
            try { 
                ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
                System.out.println("1");
                serverSocketChannel.bind(new InetSocketAddress("localhost", 8888));
                System.out.println("2");
                serverSocketChannel.configureBlocking(false);
                System.out.println("3");
                Selector selector = Selector.open();
                System.out.println("4");
                serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
                System.out.println("5");
                int keyCount = selector.select();
                System.out.println("6 keyCount=" + keyCount);
                serverSocketChannel.close();
                System.out.println("7 end!");
            } catch (IOException e) { 
                e.printStackTrace();
            }
        }

服务端输出到5，然后阻塞在select()上，  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 1]

客户端：

public static void main(String[] args) { 
            try { 
                SocketChannel socketChannel = SocketChannel.open();
                socketChannel.connect(new InetSocketAddress("localhost", 8888));
                Thread.sleep(5000);
                socketChannel.close();
            } catch (IOException e) { 
                e.printStackTrace();
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }

运行客户端：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 2]

`serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);`将OP\_ ACCEPT事件当成感兴趣的事件。因此，在运行这个客户端类后，select() 方法感知到有客户端的连接请求，服务端中的ServerSocketChannel 通道需要接受，则select()方法不再出现阻塞的效果，程序继续向下运行。

服务端不阻塞后继续向下运行，进程结束，但随后客户端也连接不到服务端了，因为服务端进程已经销毁。其实在大多数的情况下，服务端的进程并不需要销毁，因此，就要使用while(true)无限循环来无限地接受客户端的请求。但在这个过程中，有可能出现select()方法不出现阻塞的情况，造成的结果就是真正地出现“死循环”了。

#### select()方法不阻塞的原因和解决方法 ####

在某些情况下，select()是不阻塞的：

先运行服务端再运行客户端：

//循环输出
    keyCount=0
    set1 size=1
    set2 size=1
    
    keyCount=0
    set1 size=1
    set2 size=1
    
    keyCount=0
    set1 size=1
    set2 size=1

出现“死循环”的原因是在客户端连接服务端时，服务端中的通道对acceppt事件并未处理，导致accept事件一直存在，也就是select()方法一直检测到有准备好的通道要对accept事件进行处理，但一直未处理，就一直呈“死循环”输出的状态了。解决“死循环”的办法是将accept事件消化处理。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 3]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 4]

#### 出现重复消费的情况 ####

如果两个不同的通道注册到相同的选择器，那么极易出现重复消费的情况。

服务端：

public static void main(String[] args) { 
            try { 
                ServerSocketChannel serverSocketChannel1 = ServerSocketChannel.open();
                serverSocketChannel1.bind(new InetSocketAddress("localhost", 7777));
                serverSocketChannel1.configureBlocking(false);
    
                ServerSocketChannel serverSocketChannel2 = ServerSocketChannel.open();
                serverSocketChannel2.bind(new InetSocketAddress("localhost", 8888));
                serverSocketChannel2.configureBlocking(false);
    
                Selector selector = Selector.open();
                //两个注册到同一个selector
                SelectionKey selectionKey1 = serverSocketChannel1.register(selector, SelectionKey.OP_ACCEPT);
                SelectionKey selectionKey2 = serverSocketChannel2.register(selector, SelectionKey.OP_ACCEPT);
    
                boolean isRun = true;
                while (isRun) { 
                    int keyCount = selector.select();
                    Set<SelectionKey> set1 = selector.keys();
                    Set<SelectionKey> set2 = selector.selectedKeys();
                    System.out.println("keyCount=" + keyCount);
                    System.out.println("set1 size=" + set1.size());
                    System.out.println("set2 size=" + set2.size());
    
                    Iterator<SelectionKey> it = set2.iterator();
                    while (it.hasNext()) { 
                        SelectionKey key = it.next();
                        ServerSocketChannel channel = (ServerSocketChannel) key.channel();
                        SocketChannel socketChannel = channel.accept();
                        if (socketChannel == null) { 
                            System.out.println("打印这条信息证明是连接8888服务器时，重复消费的情况发生，");
                            System.out.println("将7777关联的SelectionKey对应的SocketChannel通道取出来");
                            System.out.println("但是值为null, socketChannel == null.");
                        }
                        InetSocketAddress localAddress = (InetSocketAddress) channel.getLocalAddress();
                        System.out.println(localAddress.getPort() + " 被客户端连接了!");
                    }
                }
                serverSocketChannel1.close();
                serverSocketChannel2.close();
            } catch (IOException e) { 
                e.printStackTrace();
            }
        }

客户端1：

public static void main(String[] args) { 
            try { 
                Socket socket1 = new Socket("localhost", 7777);
                socket1.close();
            } catch (IOException e) { 
                e.printStackTrace();
            }
        }

客户端2：

public static void main(String[] args) { 
            try { 
                Socket socket = new Socket("localhost", 8888);
                socket.close();
            } catch (IOException e) { 
                e.printStackTrace();
            }
        }

运行服务端和客户端1，服务端输出：

keyCount=1
    set1 size=2
    set2 size=1
    7777 被客户端连接了!

说明端口7777被客户端连接了。

再运行客户端2，控制台完整输出：

keyCount=1
    set1 size=2
    set2 size=1
    7777 被客户端连接了!
    keyCount=1
    set1 size=2
    set2 size=2
    打印这条信息证明是连接8888服务器时，重复消费的情况发生，
    将7777关联的SelectionKey对应的SocketChannel通道取出来
    但是值为null, socketChannel == null.
    7777 被客户端连接了!
    8888 被客户端连接了!

上述结果说明运行客户端2的实现代码连接服务端时，服务端对客户端2的连接请求处理过程中的set2进行第一次循环时，从SelectionKey取得的通道是绑定到7777端口上的，但本次连接的端口是8888，因此，在本次循环中执行以下代码:  
`SocketChannel socketChannel = server SocketChannel .accept() ;`  
返回的socketChannel对象的值是null。如果在后面有业务型代码，那些代码被无效地执行，下一次循环还要处理连接8888的业务，这样来看，第1次循环就是重复消费了。因此，这样是错误的，也就是出现重复无效的消费。那么内部的技术原因是什么呢?下面继续分析。

客户端2连接的端口是8888,但却重复输出“7777 被客户端连接了!”信息，造成这样的原因是变量set2在每一次循环中使用的是底层提供的同一个对象，一直在往set2里面添加已就绪的SelectionKey,一个是关联7777端口的SelectionKey,另一个是关联8888端口的SelectionKey。在这期间，从未从set2中删除SelectionKey,因此，set2 的size值为2,再使用while(iterator.hasNext())对set2循环两次，就导致了重复消费。解决重复消费问题的方法就是使用remove()方法删除set2中处理过后的SelectionKey。

#### 使用remove()方法解决重复消费问题 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 5]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 6]

`注意：set1和set2关联的各自对象是同一个`  
结论: set1 和set2一直在使用各自不变的对象，也就会出现一直向set2中添加SelectionKey造成重复消费的效果，因此，就要结合remove()方法避免重复消费。

#### int selector.select()方法返回值的含义 ####

intselector.select()方法返回值的含义是`已更新其准备就绪操作集的键的数目`，该数目可能为零或排零，非零的情况就是向set2中添加SelectionKey的个数，值为零的情况是set2中的元素并没有更改。

*  keyCount=1的含义是在已就绪的键值中添加了一个SelectionKey。
 *  set1 size=2的含义是因为有2个通道注册到了同一个选择器中，键集个数为2。
 *  set2 size=1的含义是已就绪键集中存在1个SelectionKey，这个SelectionKey就是代表7777端口对应的ServerSocketChannel。

#### 从已就绪的键集中获得通道中的数据 ####

public static void main(String[] args) { 
            try { 
                ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
                serverSocketChannel.bind(new InetSocketAddress("localhost", 8888));
                serverSocketChannel.configureBlocking(false);
                Selector selector = Selector.open();
                serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
                boolean isRun = true;
                while (isRun) { 
                    int keyCount = selector.select();
                    Set<SelectionKey> keys = selector.selectedKeys();
                    Iterator<SelectionKey> it = keys.iterator();
                    while (it.hasNext()) { 
                        SelectionKey key = it.next();
                        if (key.isAcceptable()) { 
                            ServerSocketChannel channel = (ServerSocketChannel) key.channel();
                            ServerSocket serverSocket = channel.socket();
                            Socket socket = serverSocket.accept();
                            InputStream inputStream = socket.getInputStream();
                            byte[] byteArray = new byte[1024];
                            int readLength = inputStream.read(byteArray);
                            while (readLength != -1) { 
                                System.out.println(new String(byteArray, 0, readLength));
                                readLength = inputStream.read(byteArray);
                            }
                            inputStream.close();
                            socket.close();
                            it.remove();//删除
                        }
                    }
                    serverSocketChannel.close();
                }
            } catch (IOException e) { 
                e.printStackTrace();
            }
        }
    
    
    public static void main(String[] args) { 
            try { 
                Socket socket = new Socket("localhost", 8888);
                OutputStream outputStream = socket.getOutputStream();
                for (int i = 0; i < 5; i++)
                    outputStream.write("我来自客户端！".getBytes());
                socket.close();
            } catch (IOException e) { 
                e.printStackTrace();
            }
        }

我来自客户端！我来自客户端！我来自客户端！我来自客户端！我来自客户端！

#### 对相同的通道注册不同的相关事件返回同一个SelectionKey ####

public static void main(String[] args) { 
            try { 
                ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
                serverSocketChannel.bind(new InetSocketAddress("localhost", 8888));
                serverSocketChannel.configureBlocking(false);
                Selector selector = Selector.open();
                serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
                boolean isRun = true;
                while (isRun) { 
                    int keyCount = selector.select();
                    Set<SelectionKey> set1 = selector.keys();
                    Set<SelectionKey> set2 = selector.selectedKeys();
                    System.out.println("keyCountA=" + keyCount);
                    System.out.println("set1 size=" + set1.size());
                    System.out.println("set2 size=" + set2.size());
                    System.out.println();
                    Iterator<SelectionKey> it = set2.iterator();
                    while (it.hasNext()) { 
                        SelectionKey key = it.next();
                        ServerSocketChannel channel = (ServerSocketChannel) key.channel();
                        SocketChannel socketChannel = channel.accept();
                        socketChannel.configureBlocking(false);
                        SelectionKey key2 = socketChannel.register(selector, SelectionKey.OP_READ);
                        System.out.println("key2.isRadable()=" + ((SelectionKey.OP_READ & ~key2.interestOps()) == 0));
                        System.out.println("key2.isWritable()=" + ((SelectionKey.OP_WRITE & ~key2.interestOps()) == 0));
    
                        SelectionKey key3 = socketChannel.register(selector, SelectionKey.OP_WRITE);
                        System.out.println("key3.isRadable()=" + ((SelectionKey.OP_READ & ~key2.interestOps()) == 0));
                        System.out.println("key3.isWritable()=" + ((SelectionKey.OP_WRITE & ~key3.interestOps()) == 0));
    
                        System.out.println("keyCountA=" + keyCount);
                        System.out.println("set1 size=" + set1.size());
                        System.out.println("set2 size=" + set2.size());
                        System.out.println("key2==key3结果：" + (key2 == key3));
                    }
                    Thread.sleep(Integer.MAX_VALUE);
                }
                serverSocketChannel.close();
            } catch (IOException e) { 
                e.printStackTrace();
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }

public static void main(String[] args) { 
            try { 
                Socket socket = new Socket("localhost", 8888);
                OutputStream outputStream = socket.getOutputStream();
                outputStream.write("12345".getBytes());
                socket.close();
            } catch (IOException e) { 
                e.printStackTrace();
            }
        }

服务端输出：

keyCountA=1
    set1 size=1
    set2 size=1
    
    key2.isRadable()=true
    key2.isWritable()=false
    key3.isRadable()=false
    key3.isWritable()=true
    keyCountA=1
    set1 size=2
    set2 size=1
    key2==key3结果：true

一个SocketChannel通道注册两个事件并没有创建出两个SelectionKey,而是创建出一个，read和write事件是在同一个SelectionKey中进行注册的。

另一个SelectionKey代表关联的是ServerSocketChannel通道。

#### 判断选择器是否为打开状态 ####

`public abstract boolean isOpen()`方法的作用是告知此选择器是否已打开。返回值当且仅当此选择器已打开时才返回true。  
`public abstract void close()`方法的作用是关闭此选择器。如果某个线程目前正阻塞在此选择器的某个选择方法中，则中断该线程，如同调用该选择器的wakeup()方法。所有仍与此选择器关联的未取消键已无效，其通道已注销，并且与此选择器关联的所有其他资源已释放。如果此选择器已经关闭，则调用此方法无效。关闭选择器后，除了调用此方法或wakeup()方法外，以任何其他方式继续使用它都将导致拋出ClosedSelectorException。

#### 获得SelectorProvider 对象 ####

`public abstract SelectorProvider provider()`方法的作用是返回创建此通道的提供者。

SelectorProvider provider = SelectorProvider.provider();

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70]: /images/20210813/ed39c5240cc9455cb7cbc55373ed8be6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 1]: /images/20210813/7ade36764ca740b2ad53ebcf436ad939.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 2]: /images/20210813/343b31e7c4d547f1aa5bbd0fdcaeb2ee.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 3]: /images/20210813/1a57f1d327df40cfa078c861f4533189.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 4]: /images/20210813/af8dd27f212e48b1bd1332bc64b7a75a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 5]: /images/20210813/385eae98baa14c5c92a4ac711251b6fe.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 6]: /images/20210813/7ba292602cbb45da98bfbe0bb32f4db6.png