NIO 学习笔记(二)相识篇-阿里云开发者社区

Selector 监视者轮询者选择器

Selector中的一个类，也就是监视Socket流状态的类，调用的还是操作系统提供的服务，需要配合通道和缓冲区使用。不管是多路复用还是非多路复用我们总需要一个socket，不过多路复用的时候在java中是ServerSocketChannel。

ServerSocketChannel serverSocketChannel1 = ServerSocketChannel.open();
   // 将通道设置为非阻塞，为多路复用做准备。
   serverSocketChannel1.configureBlocking(false);
   ServerSocket serverSocket = serverSocketChannel1.socket();
   // 绑定端口
   serverSocket.bind(new InetSocketAddress(8888));
复制代码

选择器呢，怎么体现哪个客户端有数据就调用对应的代码来处理客户端呢? 还是在ServerSocketChannel上，当有数据到来时我们就激活这个通道，从缓冲区中获取数据。因此我们需要让Selector来管理ServerSocketChannel，那么就是通过注册。也就是这样：

Selector selector = Selector.open();
    ServerSocketChannel serverSocketChannel1 = ServerSocketChannel.open();
    serverSocketChannel1.configureBlocking(false);
    ServerSocket serverSocket = serverSocketChannel1.socket();
    serverSocket.bind(new InetSocketAddress(8888));
    serverSocketChannel1.register(selector, SelectionKey.OP_ACCEPT);
复制代码

网络通信中JVM抽象出了四种通信状态，在SelectionKey中存放:

OP_READ 数据已经准备好了，可以读了，也成为了读就绪。
OP_WRITE 已经可以通过通道向缓冲区中写数据了，也称为写就绪。
OP_CONNECT 连接已经就绪
OP_ACCEPT 服务端已经监听到了客户端发起的连接请求，可以与服务端建立连接的。

从jdk的注释我们就可以看出，OP_ACCEPT用于服务端程序，因为一个服务端会有许多客户端，先有是否有连接的请求，再有数据是否准备完成，是否可写。OP_CONNECT用于客户端，当连接建立，我才可以向服务端写数据和读数据。

当通道上有选择器感兴趣的事件发生，比如: 连接的请求，再客户端和服务端连接建立后是可读和可写。这些事件会被选择器放入一个集合中。我们就可以循环这个集合，从客户端读取或写。一个客户端对应一些事件，这些事件不可能同时完成，总会有先后顺序，在收到客户端发起的连接请求之后，我们从中提取请求。

特别注意OP_ACCEPT只能用于选择器管理ServerSocketChannel。OP_CONNECT只能用于选择器管理SocketChannel。

聊天室服务端

介绍选择器的一部分代码就是聊天室服务端代码的一部分。我们先介绍大致的思路，在上对应的代码。

常规情况下我们还是创建用ServerSocketChannel创建Socket，然后绑定端口。用选择器管理通道，当通道上有选择器最初所感兴趣的事件之后，注意虽然我们只注册了一个对连接感兴趣的事件，但是本质上还是用的Socket，所以在服务端发现有发起建立连接的请求之后，服务端就会尝试和客户端建立连接，连接还是为了通信。

上文我们提到，在通道有选择器所感兴趣的事件之后，会将放入到一个集合中，也就是Set<SelectionKey>。通过SelectionKey，我们可以为增加选择器和通道之间的关联，最初选择器只会在通道有请求的连接时被激活，只会向集合中添加一个读就绪事件。我们可以通过SelectionKey的interestOps()为选择器添加读就绪事件和写就绪事件。然后就可以灵活的处理了，也就是说在读就绪的时候读，写就绪的时候写。

每次向选择器注册通道时就会创建一个SelectionKey键，也就是选择管理这个通道。通过调用某个SelectionKey的 cancel 方法、关闭其通道，或者通过关闭其选择器来取消该键之前，它一直保持有效。

public class ChatServer {
  // 存储客户端信息,一个客户端一个Channel   
    private static Map<String, SocketChannel> map = new HashMap<>();
    public static void main(String[] args) throws IOException {
    // 创建选择器    
        Selector selector = Selector.open();
        //创建服务端的通道
        ServerSocketChannel serverSocketChannel1 = ServerSocketChannel.open();
        // 将该通道设置为非阻塞方式
        serverSocketChannel1.configureBlocking(false);
        // 创建Socket
        ServerSocket serverSocket = serverSocketChannel1.socket();
        // 绑定端口
        serverSocket.bind(new InetSocketAddress(8888));
        // 为ServerSocketChannel注册对对客户端请求连接事件感兴趣
        // 此时该channel处于选择器得管理之下
        serverSocketChannel1.register(selector, SelectionKey.OP_ACCEPT);
        // 因为不清楚有多少客户端会连接,所以是while(true)
        while (true) {
          // 该方法将一直阻塞，知道选择器管理的通道上有选择器感兴趣的事件发生。
            selector.select();
            // 每次向选择器注册通道时就会创建一个SelectionKey键
            // 每个SelectionKey与对应的通道相连,存储通道的状态
            // 一个客户端一个SocketChannel
            Set<SelectionKey> keys = selector.selectedKeys();
            Iterator<SelectionKey> iterator = keys.iterator();
            while (iterator.hasNext()) {
                SocketChannel clietChannel = null;
                SelectionKey key = iterator.next();
                // 如果有客户端请求建立连接
                if (key.isAcceptable()) {
                 // 获取SelectionKey关联的通道
                    ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
                    // 提取请求,获取对应的socketChannel
                    clietChannel = serverSocketChannel.accept();
                    // 将处理对应的客户端信息socketChannel设置为非阻塞
                    clietChannel.configureBlocking(false);
                    // 为该通道注册读就绪事件, 选择器会询问该通道的状态,当该通道就绪时,
                    clietChannel.register(selector, SelectionKey.OP_READ);
                    // 记录处理对应客户端的channel
                    String id = "key=" + (int) (Math.random() * 9000 + 1000);
                    map.put(id, clietChannel);
                } else if (key.isReadable()) {
                  //读就绪,通过通道向缓冲区中取数据
                    clietChannel = (SocketChannel) key.channel();
                    ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                    String receiveMsg = null;
                    int result = -1;
                    try {
                        result = clietChannel.read(byteBuffer);
                    } catch (IOException e) {
                      // 正在读的时候服务端断开连接,此时数据无法继续读取
                        String clientKey = getClientKey(clietChannel);
                        System.out.println("退出了战场:" + clientKey);
                        map.remove(clientKey);
                        // 释放资源
                        clietChannel.close();
                        key.cancel();
                        continue;
                    }
                    if (result > 0) {
                        // 读之前要flip
                        byteBuffer.flip();
                        //防止乱码
                        Charset charset = Charset.forName("UTF-8");
                        receiveMsg = String.valueOf(charset.decode(byteBuffer).array());
                        System.out.println(clietChannel + ":" + receiveMsg);
                        if ("conncet".equals(receiveMsg)) {
                            receiveMsg = "新客户端上线";
                        }
                        System.out.println("读到了信息" + receiveMsg);
                      //将信息搭载在key上,为广播出去做准备。
                      // 因为我们做的是聊天室,相当于群聊。   
                        key.attach(receiveMsg);
                        // key关联通道,给key注册就相当于给通道注册。 
                        // 给对应的通道注册写事件。
                        key.interestOps(SelectionKey.OP_WRITE);
                    }
                } else if (key.isWritable()) {
                  //channel可以读可以写
                    clietChannel = (SocketChannel) key.channel();
                    // 获取对应的客户端
                    String name = getClientKey(clietChannel);
                    for (Map.Entry<String, SocketChannel> entrySet : map.entrySet()) {
                        SocketChannel entryValue = entrySet.getValue();
                        ByteBuffer broadCast = ByteBuffer.allocate(1024);
                        broadCast.put((name + ":" + key.attachment()).getBytes());
                        broadCast.flip();
                        //将缓冲区的东西发送给每一个通道
                        entryValue.write(broadCast);
                    }
                    // 给key关联的通道注册读事件。
                    key.interestOps(SelectionKey.OP_READ);
                } else {
            System.out.println("hello world");
                }
            }
            //清空
            keys.clear();
        }
    }
    private static String getClientKey(SocketChannel clietChannel) {
        String key = null;
        Set<Map.Entry<String, SocketChannel>> entrySet = map.entrySet();
        for (Map.Entry<String, SocketChannel> value : entrySet) {
            if (value.getValue() == clietChannel) {
                key = value.getKey();
                break;
            }
        }
        return key;
    }
}
复制代码

为什么清空keys?

为什么最后清空 Set<SelectionKey>？我清空了还怎么用?让我来解释一下，事实上选择器维护了三张表(三个集合)。

注册表: 当我们为通道注册事件，也就是将该通道纳入到选择器得管辖范围之内时，也就是channel.register。就会产生一个对应的key进入这张表。只有当我们调用key.cancel()方法时,这个key才会被移除。通过Selector类中的keys()返回这个集合。
已就绪表: 当我们调用selector.select()方法,这个选择器进入注册表查找,查看哪个通道上的事件是已经就绪了,然后将这些可就绪的key(就是selectionKey,通过selectionKey获取所关联通道的状态,即通道上的事件是否就绪),这个表不会被选择器清空,即使我们再次调用selector.select(),他也不会清空这张表中已存在的key。
通道脱离管辖表: 也就是说selectionKey调用cancle()方法,那么该selectionKey所对应的通道将会脱离选择器的管辖。

这就是我们最后调用清空keys集合的原因,如果遍历该集合之后，所有就绪的事件都已经处理而不清空，那么下次就绪的key我们就获取不到。因为set是不可重复的。每次执行完selector.select(),选择器都会轮询被管理的通道，将通道上感兴趣的事件就绪的通道所对应的selectionKey加入到已就绪的表中。

为什么在写事件就绪之后,还要为通道注册读事件

我们要聊聊写事件,写事件就绪的条件很容易满足,就是操作系统给对应socket分配的缓冲区还有空闲,一般情况下都要满足。 key.interestOps(SelectionKey.OP_READ)相当于给该通道注册读事件,等价于取消掉key关联通道对写事件感兴趣。因为一个key目前只能对一个事件感兴趣。如果你不取消对写事件感兴趣,那么写事件会不断的被触发。

聊天室客户端

public class ChatClient {
    public static void main(String[] args) {
        try {
            //客户端用socketchannel
            SocketChannel socketChannel = SocketChannel.open();
            // 将该通道设置为非阻塞
            socketChannel.configureBlocking(false);
          // 创建选择器    
            Selector selector = Selector.open();
            // 确定服务端
            socketChannel.connect(new InetSocketAddress("127.0.0.1",8888));
            // 将该通道纳入选择器得管理范围之下。客户端只有跟服务端建立连接,因此对连接感兴趣就好
            socketChannel.register(selector, SelectionKey.OP_CONNECT);
            while (true){
             // 轮询通道的感兴趣事件是否就绪,有就绪事件即返回。没有就一直陷入阻塞
               selector.select();
             // 获取就绪的通道   
               Set<SelectionKey> selectKeys = selector.selectedKeys();
                Iterator<SelectionKey> keyIterator = selectKeys.iterator();
                while(keyIterator.hasNext() ){
                    SelectionKey selectKey = keyIterator.next();
                    // 判断连接是否完成
                    if (selectKey.isConnectable()){
                        ByteBuffer sendBuffer = ByteBuffer.allocate(1024);
                        SocketChannel clientChannel = (SocketChannel)selectKey.channel();
                        // 首次连接完成后,程序内部仍然需要一些处理，判断处理是否完成。
                        // 这一行可以去掉,只是为了说明这个方法。未完成连接时返回false
                        if (clientChannel.isConnectionPending()){
                          // 只完成一次,当通道完成连接后,返回为true
                            if (clientChannel.finishConnect()) {
                                System.out.print("连接服务端成功");
                                sendBuffer.put("conncet".getBytes());
                                sendBuffer.flip();
                                clientChannel.write(sendBuffer);
                            }
                        }
                        // 启动一个线程监听用户的输入,在目前这种模式下。用户随时可以写。随时发送信息
                        new Thread(()->{
                            while (true){
                                sendBuffer.clear();
                                InputStreamReader reader = new InputStreamReader(System.in);
                                BufferedReader bufferReader = new BufferedReader(reader);
                                try {
                                    String message = bufferReader.readLine();
                                    sendBuffer.put(message.getBytes());
                                    sendBuffer.flip();
                                    clientChannel.write(sendBuffer);
                                } catch (IOException e) {
                                    e.printStackTrace();
                                }
                            }
                        }).start();
                        // 连接就绪,注册对读事件感兴趣
                        clientChannel.register(selector,SelectionKey.OP_READ);
                    }else if(selectKey.isReadable()){
                        ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                        SocketChannel clientChannel = (SocketChannel)selectKey.channel();
                        int len = clientChannel.read(readBuffer);
                        if (len > 0){
                            String receive = new String(readBuffer.array(), 0, len);
                            System.out.println(receive);
                        }
                    }
                }
                selectKeys.clear();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
复制代码