1、选择器介绍

选择器的作用是完成IO的多路复用

并不是所有通道都可以被选择器监控或者选择的，例如FileChannel就不能被选择器复用。判断一个通道是否能被选择器监控或者选择只需看其是否继承了抽象类SelectableChannel

选择器和通道是监控和被监控的关系。通过选择器可以监控多个通道的IO情况。
一般来说，一个单线程处理一个选择器，一个选择器可以监控很多通道，通过选择器，一个单线程就可以处理数千数万甚至更多的通道。在极端情况下（数万连接），只用一个线程就可以处理所有的通道，会大量的减少线程之间上下文切换的开销。

单线程可以配合 Selector 完成对多个 Channel 可读写事件的监控，这称之为多路复用

  • 多路复用仅针对网络 IO、普通文件 IO 没法利用多路复用
  • 如果不用 Selector 的非阻塞模式，线程大部分时间都在做无用功，而 Selector 能够保证
    • 有可连接事件时才去连接
    • 有可读事件才去读取
    • 有可写事件才去写入
      • 限于网络传输能力，Channel 未必时时可写，一旦 Channel 可写，会触发 Selector 的可写事件

通道和选择器的关系通过通道注册的形式确定，调用通道的注册方法可以将通道实例注册到一个选择器中，注册方法有两个参数一个是指定的选择器实例，一个是指定选择器监控的IO事件类型。

  • 可读
  • 可写
  • 连接
  • 接收

2、选择器使用

2.1、监听 Channel 事件

可以通过下面三种方法来监听是否有事件发生，方法的返回值代表有多少 channel 发生了事件
方法1，阻塞直到绑定事件发生
int count = selector.select();
方法2，阻塞直到绑定事件发生，或是超时（时间单位为 ms）
int count = selector.select(long timeout);
方法3，不会阻塞，也就是不管有没有事件，立刻返回，自己根据返回值检查是否有事件
int count = selector.selectNow();

💡 select 何时不阻塞

  • 事件发生时
    • 客户端发起连接请求，会触发 accept 事件
    • 客户端发送数据过来，客户端正常、异常关闭时，都会触发 read 事件，另外如果发送的数据大于 buffer 缓冲区，会触发多次读取事件
    • channel 可写，会触发 write 事件
    • 在 linux 下 nio bug 发生时
  • 调用 selector.wakeup()
  • 调用 selector.close()
  • selector 所在线程 interrupt

2.2、注册选择器

SelectionKey serverSelectKey = ssc.register(通道, 关注事件, 关联的buffer);

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

通过或运算监控通道的多个事件：
int ket = SelectionKey.OP_READ | SelectionKey.OP_WRITE；

3、选择器实战

server端

  1. @Slf4j
  2. public class SelectorServer {
  4.     public static void main(String[] args) throws IOException {
  6.         //1、创建selector 选择器,管理多个channel
  7.         Selector selector = Selector.open();
  8.         //2、创建服务器
  9.         ServerSocketChannel ssc = ServerSocketChannel.open();
  10.         ssc.configureBlocking(false);
  11.         ssc.bind(new InetSocketAddress(8989));
  12.         //3、注册selector
  13.         /**
  14.          * SelectionKey为事件发生后，通过它可以知道事件和那个channel发生的事件
  15.          * 事件分为：accept(有连接请求时触发),connect(建立连接时触发),read,write
  16.          */
  17.         SelectionKey serverSelectKey = ssc.register(selector, 0, null);
  18.         //服务端只关注accept
  19.         serverSelectKey.interestOps(SelectionKey.OP_ACCEPT);
  20.         log.info("register key : {}",serverSelectKey);
  23.         while (true){
  24.             //3、select 没有事件发生则线程阻塞,在事件未处理时，它不会阻塞，事件处理或者取消就会阻塞
  25.             // selector会在事件发生后向selectedKeys集合中加入key,但是不会删除key，所以处理完一个key,需要删除
  26.             selector.select();
  27.             //4、处理事件
  28.             Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
  29.             while(iterator.hasNext()){
  30.                 SelectionKey key = iterator.next();
  31.                 //处理key时要从selectedKeys集合删除
  32.                 iterator.remove();
  33.                 log.info("key: {}",key);
  34.                 //5、区分事件类型
  35.                 if(key.isAcceptable()){
  36.                     ServerSocketChannel serverChannel = (ServerSocketChannel) key.channel();
  37.                     SocketChannel clientChannel = serverChannel.accept();
  38.                     log.info("accepted... {}",clientChannel);
  39.                     clientChannel.configureBlocking(false);
  40.                     ByteBuffer buffer = ByteBuffer.allocate(1024);
  41.                     //byteBuffer非线程安全，将byteBuffer作为附件关联到clientSelectKey上
  42.                     SelectionKey clientSelectKey = clientChannel.register(selector, SelectionKey.OP_READ, buffer);
  43.                     log.info("clientSelectKey {}",clientSelectKey);
  44.                 }else if(key.isReadable()){
  45.                     try {
  46.                         SocketChannel clientChannel = (SocketChannel) key.channel();
  47.                         //获取关联的byteBuffer
  48.                         ByteBuffer buffer = (ByteBuffer) key.attachment();
  49.                         //如果客户端是正常断开，read返回-1
  50.                         int read = clientChannel.read(buffer);
  51.                         if(read == -1){
  52.                             key.cancel();
  53.                             log.info("客户端 {} 断开",clientChannel);
  54.                         }else {
  55.                             split(buffer);
  56.                             //扩容
  57.                             if(buffer.limit() == buffer.position()){
  58.                                 ByteBuffer newBuffer = ByteBuffer.allocate(buffer.capacity() * 2);
  59.                                 buffer.flip();
  60.                                 newBuffer.put(buffer);
  61.                                 key.attach(newBuffer);
  62.                             }
  63.                         }
  64.                     } catch (IOException e) {
  65.                         //避免客户端异常断开造成异常导致服务器停掉,需要将key从选择器key集合中删除
  66.                         e.printStackTrace();
  67.                         key.cancel();
  68.                     }
  69.                 }
  70.             }
  71.         }
  72.     }
  74.     private static void split(ByteBuffer source) {
  75.         source.flip();
  76.         for (int i = 0; i < source.limit() ; i++) {
  77.             if (source.get(i) == '#'){
  78.                 int len = i + 1 - source.position();
  79.                 ByteBuffer target = ByteBuffer.allocate(len);
  80.                 for (int j = 0; j < len; j++) {
  81.                     byte b = source.get();
  82.                     target.put(b);
  83.                 }
  84.                 target.flip();
  85.                 log.info(StandardCharsets.UTF_8.decode(target).toString());
  86.             }
  87.         }
  88.         source.compact();
  89.     }
  90. }

💡 事件发生后能否不处理

事件发生后，要么处理，要么取消（cancel），不能什么都不做，否则下次该事件仍会触发，这是因为 nio 底层使用的是水平触发

💡 为何要 iter.remove()

因为 select 在事件发生后，就会将相关的 key 放入 selectedKeys 集合，但不会在处理完后从 selectedKeys 集合中移除，需要我们自己编码删除。例如

  • 第一次触发了 ssckey 上的 accept 事件，没有移除 ssckey
  • 第二次触发了 sckey 上的 read 事件，但这时 selectedKeys 中还有上次的 ssckey ，在处理时因为没有真正的 serverSocket 连上了，就会导致空指针异常

💡 cancel 的作用

cancel 会取消注册在 selector 上的 channel，并从 keys 集合中删除 key 后续不会再监听事件

💡 客户端断开事件

如果客户端是正常断开，read返回-1 客户端异常断开会抛出IOException异常

client端

  1. public class Client {
  2.     public static void main(String[] args) throws IOException {
  3.         SocketChannel sc = SocketChannel.open();
  4.         sc.connect(new InetSocketAddress("localhost",8989));
  5.         Scanner scanner = new Scanner(System.in);
  6.         while (scanner.hasNext()) {
  7.             String str = scanner.next();
  8.             if("close".equals(str)){
  9.                 sc.close();
  10.             }else {
  11.                 sc.write(StandardCharsets.UTF_8.encode(str));
  12.             }
  13.         }
  15.     }
  16. }