JavaNIO三组件SelecotrChannel实现原理解析

　　一、NIO聊天室入门案例
　　在学习原理之前，先来了解一个Java NIO实现聊天室的小案例，该案例只有三个类：NioServer 聊天室服务端、NioClient 聊天室客户端、ClientThread 客户端线程。  服务端代码： /**  * Nio聊天室服务端  *  * @author csp  * @date 2021-11-30 4:13 下午  */ public class NioServer {      /**      * 聊天室成员列表：      */     Map memberChannels;      /**      * 端口      */     private static final int PORT = 8080;      /**      * 选择器      */     private Selector selector;      /**      * 管道      */     private ServerSocketChannel server;      /**      * 缓冲      */     private ByteBuffer buffer;      public NioServer() throws IOException {         // 初始化Selector选择器         this.selector = Selector.open();         // 初始化Channel通道         this.server = getServerChannel(selector);         // 初始化Buffer缓冲：         this.buffer = ByteBuffer.allocate(1024);         // 初始化聊天室成员列表         memberChannels = new ConcurrentHashMap<>();     }      /**      * 初始化Channel通道      *      * @param selector      * @return      * @throws IOException      */     private ServerSocketChannel getServerChannel(Selector selector) throws IOException {         // 开辟一个Channel通道         ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();         // 通道设置为非阻塞模式         serverSocketChannel.configureBlocking(false);         // 通道注册绑定Selector选择器，通道中数据的事件类型为OP_ACCEPT         serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);         // 通道绑定端口         serverSocketChannel.socket().bind(new InetSocketAddress(PORT));          return serverSocketChannel;     }      /**      * 事件监听      */     public void listen() throws IOException {         System.out.println(＂服务端启动......＂);         try {             // 无限循环             while (true) {                 // 作用：至少需要有一个事件发生，否则（如果count == 0）就继续阻塞循环                 int count = selector.select();                 if (count == 0) {                     continue;                 }                 // 获取SelectorKey的集合                 Set keySet = selector.selectedKeys();                  Iterator iterator = keySet.iterator();                 while (iterator.hasNext()) {                     // 当前事件对应的SelectorKey                     SelectionKey key = iterator.next();                     // 删除当前事件：表示当前事件已经被消费了                     iterator.remove();                     if (key.isAcceptable()) {                         // Accept类型事件：                         // 通过key获取ServerSocketChannel                         ServerSocketChannel server = (ServerSocketChannel) key.channel();                         // 通过ServerSocketChannel获取SocketChannel                         SocketChannel channel = server.accept();                         // channel设置为非阻塞模式                         channel.configureBlocking(false);                         // channel绑定选择器                         channel.register(selector, SelectionKey.OP_READ);                          // 从channel中获取Host、端口等信息                         System.out.println(＂客户端连接：＂                                 + channel.socket().getInetAddress().getHostName() + ＂:＂                                 + channel.socket().getPort());                     } else if (key.isReadable()) {                         // Read类型事件                         SocketChannel channel = (SocketChannel) key.channel();                         // 用于解密消息内容                         CharsetDecoder decoder = Charset.forName(＂UTF-8＂).newDecoder();                          // 将消息数据从通道channel读取到缓冲buffer                         //ByteBuffer buffer = ByteBuffer.allocate(50);                         buffer.clear();                         channel.read(buffer);                         buffer.flip();                         // 获取解密后的消息内容：                         String msg = decoder.decode(buffer).toString();                         if (!＂＂.equals(msg)) {                             System.out.println(＂收到：＂ + msg);                             if (msg.startsWith(＂username=＂)) {                                 String username = msg.replaceAll(＂username=＂, ＂＂);                                 memberChannels.put(username, channel);                                 System.out.println(＂用户总数：＂ + memberChannels.size());                             } else {                                 // 转发消息给客户端                                 String[] arr = msg.split(＂:＂);                                 if (arr.length == 3) {                                     // 发送者                                     String from = arr[0];                                     // 接收者                                     String to = arr[1];                                     // 发送内容                                     String content = arr[2];                                     System.out.println(from + ＂发送给＂ + to + ＂的消息：＂ + content);                                      if (memberChannels.containsKey(to)) {                                         // 解密                                         CharsetEncoder encoder = Charset.forName(＂UTF-8＂).newEncoder();                                         // 给接收者发送消息                                         memberChannels.get(to).write(encoder.encode(CharBuffer.wrap(from + ＂:＂ + content)));                                     }                                 }                             }                         }                      }                 }             }         }catch (Exception e){             System.out.println(＂服务端启动失败......＂);             e.printStackTrace();         }finally {             try {                 // 先关闭选择器，在关闭通道                 // 调用 close() 方法将会关闭Selector，同时也会将关联的SelectionKey失效，但不会关闭Channel。                 selector.close();                 server.close();             } catch (IOException e) {                 e.printStackTrace();             }         }     }      public static void main(String[] args) throws IOException {         // 服务端启动：         new NioServer().listen();     } } 客户端线程类： /**  * Nio聊天室客户端线程  *  * @author csp  * @date 2021-11-30 4:13 下午  */ public class ClientThread extends Thread {     /**      * 解密      */     private CharsetDecoder decoder = Charset.forName(＂UTF-8＂).newDecoder();      /**      * 加密      */     private CharsetEncoder encoder = Charset.forName(＂UTF-8＂).newEncoder();      /**      * 选择器      */     private Selector selector = null;      /**      * 通道      */     private SocketChannel socket = null;      /**      * 通道key      */     private SelectionKey clientKey = null;      /**      * 用户名      */     private String username;      public ClientThread(String username) {         try {             // 创建一个Selector             selector = Selector.open();              // 创建Socket并注册             socket = SocketChannel.open();             socket.configureBlocking(false);             clientKey = socket.register(selector, SelectionKey.OP_CONNECT);              // 连接到远程地址             InetSocketAddress ip = new InetSocketAddress(＂localhost＂, 8080);             socket.connect(ip);              this.username = username;         } catch (IOException e) {             e.printStackTrace();         }     }      /**      * 开辟读取事件的线程      */     @Override     public void run() {         try {             // 监听事件（无限循环）             while (true) {                 // 监听事件                 selector.select();                 // 事件来源列表                 Iterator it = selector.selectedKeys().iterator();                 while (it.hasNext()) {                     SelectionKey key = it.next();                     // 删除当前事件                     it.remove();                      // 判断事件类型                     if (key.isConnectable()) {                         // 连接事件                         SocketChannel channel = (SocketChannel) key.channel();                         if (channel.isConnectionPending())                             channel.finishConnect();                         channel.register(selector, SelectionKey.OP_READ);                         System.out.println(＂连接服务器端成功！＂);                          // 发送用户名                         send(＂username=＂ + this.username);                     } else if (key.isReadable()) {                         // 读取数据事件                         SocketChannel channel = (SocketChannel) key.channel();                          // 读取数据                         ByteBuffer buffer = ByteBuffer.allocate(50);                         channel.read(buffer);                         buffer.flip();                         String msg = decoder.decode(buffer).toString();                         System.out.println(＂收到：＂ + msg);                     }                 }             }         } catch (IOException e) {             e.printStackTrace();         } finally {             // 关闭             try {                 selector.close();                 socket.close();             } catch (IOException e) {             }         }     }      /**      * 发送消息      *      * @param msg      */     public void send(String msg) {         try {             SocketChannel client = (SocketChannel) clientKey.channel();             client.write(encoder.encode(CharBuffer.wrap(msg)));         } catch (Exception e) {             e.printStackTrace();         }     }      /**      * 关闭客户端      */     public void close() {         try {             selector.close();             socket.close();         } catch (IOException e) {         }     } } 客户端代码：/**  * Nio聊天室客户端  *  * @author csp  * @date 2021-12-09 17:03:33  */ public class NioClient {     public static void main(String[] args) {         // 当前客户端的用户名         String username = ＂lufei＂;         // 为当前客户端开辟一个线程         ClientThread client = new ClientThread(username);         client.start();          // 输入输出流         BufferedReader sin = new BufferedReader(new InputStreamReader(System.in));          try {             // 循环读取键盘输入             String readline;             while ((readline = sin.readLine()) != null) {                 if (readline.equals(＂bye＂)) {                     client.close();                     System.exit(0);                 }                 // 发送消息                 client.send(username + ＂:＂ + readline);             }         } catch (IOException e) {             e.printStackTrace();         }     } } 运行测试：
　　启动运行测试一下效果！
　　服务端先启动，控制台打印：  服务端启动......
　　接着启动客户端，控制台打印：  连接服务器端成功！
　　这时候服务端会打印客户端的连接信息以及用户名等信息：
　　图片
　　测试客户端向服务的发送消息，客户端控制台输入  Hello 我是lufei!  ，这时候服务端会收到发送过来的消息内容：
　　图片
　　我们可以再建立一个客户端启动类NioClient2，并将其启动，服务端会收到客户端2的消息：
　　图片
　　让客户端1和客户端2之间发送消息：
　　图片
　　图片
　　服务端控制台打印：
　　图片
　　这样，一个简单的聊天室就搭建成功了，如果小伙伴想自行完善，可以把代码拷贝一下，自己去设计自己想要实现的聊天室功能。
　　熟悉了NIO通信的小案例之后，我们通过一张图来分析一下其实现原理：
　　从图中可以看出，当有读或写等任何注册的事件发生时，可以从 Selector中获得相应的SelectionKey，同时从SelectionKey中可以找到发生的事件和该事件所发生的具体的SelectableChannel，以获得客户端发送过来的数据。  二、Selector 选择器1、Selector 继承体系
　　NIO中实现非阻塞 I/O的核心对象是Selector，Selector是注册各种I/O事件的地方，而且当那些事件发生时，就是Seleetor告诉我们所发生的事件。
　　使用NIO中非阻塞I/O编写服务器处理程序，大体上可以分为下面三个步骤：  (1) 向Selector对象注册感兴趣的事件。  (2) 从Selector中获取感兴趣的事件。  (3) 根据不同的事件进行相应的处理。  2、Selector 选择器的创建
　　在聊天室案例的NioServer服务端类中，选择器的初始化创建位于其构造函数中：  public NioServer() throws IOException {     // 初始化Selector选择器     // 也可以通过实现java.nio.channels.spi.SelectorProvider.openSelector()抽象方法自定义一个Selector。     this.selector = Selector.open();     // 初始化Channel通道     // 初始化Buffer缓冲：     // 初始化聊天室成员列表 }
　　Selector  可以通过它自己的open()  方法创建，借助java.nio.channels.spi.SelectorProvider  类创建一个新的 Selector 选择器。也可以通过实现java.nio.channels.spi.SelectorProvider  类的抽象方法openSelector()  来自定义实现一个Selector。Selector 一旦创建将会一直处于 open 状态直到调用了close()  方法为止。
　　我们跟进这个  open()  方法：public abstract class Selector implements Closeable {        protected Selector() {}        // 该方法返回一个Selector选择器：     public static Selector open() throws IOException {        // 通过SelectorProvider构建选择器         return SelectorProvider.provider().openSelector();     }    ... }
　　继续向下跟进  SelectorProvider.provider()  方法：// 该方法位于SelectorProvider类中吗，可以获取能够构建Selector选择器的SelectorProvider对象 public static SelectorProvider provider() {     synchronized (lock) {         if (provider != null)             return provider;         return AccessController.doPrivileged(             new PrivilegedAction() {                 public SelectorProvider run() {                         if (loadProviderFromProperty())                             return provider;                         if (loadProviderAsService())                             return provider;                      // DefaultSelectorProvider会根据不同的操作系统去创建不同的SelectorProvider                         provider = sun.nio.ch.DefaultSelectorProvider.create();                         return provider;                     }                 });     } }
　　我们跟进  DefaultSelectorProvider.create()  方法，它会根据不同的操作系统去创建不同的SelectorProvider：如果是Windows操作系统，则创建的是  WindowsSelectorProvider  对象。  如果是MacOS操作系统，则创建的是  KQueueSelectorProvider  对象。  如果是Linux操作系统，则创建的是  EPollSelectorProvider  对象。
　　例如我使用的是Mac系统，那么跟进  create()  方法时，进入如下代码：public class DefaultSelectorProvider {     private DefaultSelectorProvider() {     }      public static SelectorProvider create() {         return new KQueueSelectorProvider();     } }  // 继续往下跟进KQueueSelectorProvider()，进入KQueueSelectorProvider类： public class KQueueSelectorProvider extends SelectorProviderImpl {     public KQueueSelectorProvider() {     }      public AbstractSelector openSelector() throws IOException {         // KQueueSelectorImpl是具体实现类，我们继续往下跟进，去看看实现逻辑：         return new KQueueSelectorImpl(this);     } }
　　继续跟进，进入KQueueSelectorImpl类内部：  class KQueueSelectorImpl extends SelectorImpl {     // 用于存放read端的文件描述     protected int fd0;     // 用于存放write端的文件描述     protected int fd1;          ......        KQueueSelectorImpl(SelectorProvider var1) {         super(var1);        // makePipe方法是native修饰的本地方法，其作用是返回两个文件描述符var2，         // var2高位存放的是通道read端的文件描述符，低32位存放的是write端的文件描述符。         long var2 = IOUtil.makePipe(false);         // 文件描述符fd0只读数据         this.fd0 = (int)(var2 >>> 32);         // 文件描述符fd1只写数据         this.fd1 = (int)var2;                  ......     }        ...... }
　　问题：为什么在不同操作系统平台，Provider不同呢？
　　因为网络IO是跟操作系统息息相关的，不同的操作系统的实现可能都不一样。比如我们在Linux操作系统安装的JDK版本，和Windows操作系统上就不太一样。  3、Selector 选择器绑定 Channel 管道
　　如聊天室案例中服务端 NioServer，Channel 管道与 Selector 选择器的绑定通过如下方式：  /**  * 初始化Channel通道  *  * @param selector  * @return  * @throws IOException  */ private ServerSocketChannel getServerChannel(Selector selector) throws IOException {     // 开辟一个Channel通道     ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();     // 通道设置为非阻塞模式     serverSocketChannel.configureBlocking(false);     // 为了将Channel跟Selector绑定在一起，我们需要将Channel注册到Selector上，调用Channel的register()方法     // 通道中数据的事件类型为OP_ACCEPT（事件类型总共有4种，详情请参考下文的第4小节，管道与选择器之间的桥梁 SelectionKey ）     serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);     // 通道绑定端口     serverSocketChannel.socket().bind(new InetSocketAddress(PORT));     return serverSocketChannel; }
　　注意  ：Channel必须是非阻塞模式才能注册到Selector上，所以，无法将一个FileChannel注册到Selector，因为FileChannel没有所谓的阻塞还是非阻塞模式。
　　管道 Channel 和 选择器 Selector 的关系  ：  Selector 通过不断轮询的方式同时监听多个 Channel 的事件，注意，这里是同时监听，一旦有 Channel 准备好了，它就会返回这些准备好了的 Channel，交给处理线程去处理。  在NIO编程中，通过 Selector 我们就实现了一个线程同时处理多个连接请求的目标，也可以一定程序降低服务器资源的消耗。  4、管道与选择器之间的桥梁 SelectionKey
　　我们再来看一下 Selector 与 Channel 的关系图：
　　图片
　　如上图所示，将管道与注册到选择器上时，  register()  方法需要传递2个参数，一个是 Selector 选择器对象，另一个是管道中事件的类型 SelectionKey，选择器通过该对象的静态变量值去识别不同管道中的事件内容。所以SelectionKey  又可以看作是Channel和Selector之间的一座桥梁，把两者绑定在了一起。// 为了将Channel跟Selector绑定在一起，我们需要将Channel注册到Selector上，调用Channel的register()方法 // 通道中数据的事件类型为OP_ACCEPT serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
　　SelectionKey 有如下4种事件类型：  /**  * 读事件：0000 0001  */ public static final int OP_READ = 1 << 0;  /**  * 写事件：0000 0100  */ public static final int OP_WRITE = 1 << 2;  /**  * 连接事件：0000 1000，连接操作，Client端支持的一种操作  */ public static final int OP_CONNECT = 1 << 3;  /**  * 接受事件：0001 0000，可接受操作仅ServerSocketChannel支持  */ public static final int OP_ACCEPT = 1 << 4;  5、SelectionKey 的几个常用方法
　　参考文章：https://juejin.cn/post/6844903440573792270  5.1、interestOps()方法作用：返回代表需要Selector监控的IO操作的，可以通过以下方法来判断Selector是否对Channel的某种事件感兴趣。  interest数据集：当前Channel感兴趣的操作，此类操作将会在下一次选择器   select()   操作时被交付，可以通过selectionKey.interestOps(int)  进行方法修改。使用方式如下：  // 获取该selectionKey感兴趣的事件集 int interestSet = selectionKey.interestOps();   boolean isInterestedInAccept =     (interestSet & SelectionKey.OP_ACCEPT) == SelectionKey.OP_ACCEPT； boolean isInterestedInConnect = interestSet & SelectionKey.OP_CONNECT; boolean isInterestedInRead = interestSet & SelectionKey.OP_READ; boolean isInterestedInWrite = interestSet & SelectionKey.OP_WRITE; 5.2、readyOps()方法作用：获取此selectionKey键上的ready操作集合，即在当前通道上已经就绪的事件。可以通过  readyOps()  方法获取所有就绪了的事件，也可以通过isXxxable()  方法检查某个事件是否就绪。ready数据集：表示此选择键上，已经就绪的操作，每次   select()   时，选择器都会对ready集合进行更新，外部程序无法修改此集合。使用方式如下：   // 创建ready集合的方法 int readySet = selectionKey.readyOps(); // 检查这些操作是否就绪的方法 boolean isReadable();// 是否可读，是返回 true，检测此键是否为＂read＂事件.等效于:k.,readyOps() & OP_READ != 0; boolean isWritable()：// 是否可写，是返回 true boolean isConnectable()：// 是否可连接，是返回 true boolean isAcceptable()：// 是否可接收，是返回 true 5.3、channel()、selector()方法作用：通过  channel()、selector()  方法可以获取绑定的通道Channel和选择器Selector，代码案例如下：Channel channel = selectionKey.channel();Selector selector = selectionKey.selector();  5.4、attachment()方法
　　可以将一个或者多个附加对象绑定到SelectionKey上，以便容易的识别给定的通道。通常有两种方式：  在注册的时候直接绑定：    SelectionKey key=channel.register(selector,SelectionKey.OP_READ,theObject); 在绑定完成之后附加：  selectionKey.attach(theObject);// 绑定
　　绑定之后，可通过对应的SelectionKey取出该对象：  selectionKey.attachment();
　　如果要取消该对象，则可以通过该种方式：  selectionKey.attach(null)
　　需要注意的是如果附加的对象不再使用，一定要人为清除，因为垃圾回收器不会回收该对象，若不清除的话会成内存泄漏。
　　一个单独的通道可被注册到多个选择器中，有些时候我们需要通过  isRegistered（）  方法来检查一个通道是否已经被注册到任何一个选择器上。通常来说，我们并不会这么做。
　　有点类似于ThreadLocal，可以让不同的线程都拥有一份自己的变量副本，且相互隔离，但是区别在于ThreadLocal的对象过期后会被自动回收的！  6、Selector 的几个常用方法这一部分参考自这 2 篇文章：
　　https://blog.csdn.net/tangtong1/article/details/103414499https://blog.csdn.net/weixin_34237596/article/details/92738331  6.1、select()方法
　　一旦将一个或多个Channel注册到Selector上了，我们就可以调用它的  select()  方法了，它会返回注册时感兴趣的事件中就绪的事件。
　　select()  方法有三种变体：select()  ，无参数，阻塞到至少有一个通道在你注册的事件上就绪了才返回。（当然是我们注册的感兴趣的事件）。select(timeout)  ，带超时，阻塞直到某个Channel有就绪的事件了，或者超时了才返回。selectNow()  ，立即返回，非阻塞，只要有通道就绪就立刻返回。
　　select()  方法返回的 int 值表示有多少通道已经就绪，是自上次调用select()  方法后有多少通道变成就绪状态。之前在select()  调用时进入就绪的通道不会在本次调用中被记入，而在前一次select()  调用进入就绪但现在已经不在处于就绪的通道也不会被记入。例如：首次调用select()  方法，如果有一个通道变成就绪状态，返回了1，若再次调用select()  方法，如果另一个通道就绪了，它会再次返回1。如果对第一个就绪的Channel没有做任何操作，现在就有两个就绪的通道，但在每次select()  方法调用之前，只有一个通道就绪了。
　　一旦调用  select()  方法，并且返回值不为 0 时，则可以通过调用Selector的selectedKeys()  方法来访问已选择键集合。如下：Set selectionKeySet = selector.selectedKeys(); 6.2、selectKeys()方法
　　Selector.selectKeys()  ，可以获取该选择器相关联的SelectionKey集合，通过遍历这些SelectorKey，可以进一步获取其感情兴趣的事件类型，以及其关联的Channel通道。Set selectedKeys = selector.selectedKeys();  Iterator it = selectedKeys.iterator();  while(it.hasNext()) {     SelectionKey key = keyIterator.next();      if(key.isAcceptable()) {         // 可连接事件     } else if (key.isConnectable()) {         // 连接事件     } else if (key.isReadable()) {         // 读取数据事件     } else if (key.isWritable()) {         // 写入事件     }     it.remove(); }
　　最后，一定要记得调用  it.remove();  移除已经处理的 SelectionKey。6.3、wakeUp()方法
　　前面我们说了调用  select()  方法时，调用者线程会进入阻塞状态，直到有就绪的 Channel 才会返回。其实也不一定，wakeup()  就是用来破坏规则的，可以在另外一个线程调用wakeup()  方法强行唤醒这个阻塞的线程，这样select()  方法也会立即返回。
　　如果调用  wakeup()  时并没有线程阻塞在select()  上，那么，下一次调用select()  将立即返回，不会进入阻塞状态。这跟LockSupport.unpark()  方法是比较类似的。6.4、close()方法
　　调用   close()  方法将会关闭 Selector，同时也会将关联的 SelectionKey 失效，但不会关闭 Channel。三、Channel 通道1、Channel 继承体系
　　图片
　　通道是一个对象，通过它可以读取和写入数据，当然所有数据都通过  Buffer  对象来处理。  我们永远不会将字节直接写入通道，而是将数据写入包含一个或者多个字节的缓冲区。同样也不会直接从通道中读取字节，而是将数据从通道读入缓冲区，再从缓冲区获取这个字节  。
　　使用NIO读取数据可以分为下面三个步骤：  (1)从FileInputStream获取Channel。  (2)创建Buffer。  (3)将数据从Channel读取到Buffer中。
　　使用NIO写入数据同样分为三个步骤：  (1)从FileInputStream获取Channel。  (2)创建Buffer。  (3)将数据从Channel写入Buffer。  2、Channel 与 Steam 流的区别
　　参考自：https://blog.csdn.net/tangtong1/article/details/103341597
　　BIO是面向流（Stream）编程的，流又分成 InputStream 和 OutputStream ，那么 Channel 和 Stream 有什么区别呢？  Channel 可以同时支持读和写，而 Stream 只能支持单向的读或写  （所以分成InputStream和OutputStream）。  Channel支持异步读写，Stream通常只支持同步  。  Channel总是读向（read into）Buffer，或者写自（write from）Buffer（有点绕，以 Channel 为中心，从 Channel 中读出数据到 Buffer，从 Buffer 中往 Channel 写入数据），可以参考如下代码：
　　将 Channel 管道中的数据读取到 Buffer：  if (key.isReadable()) {    // 读取数据事件    SocketChannel channel = (SocketChannel) key.channel();     // 读取数据    ByteBuffer buffer = ByteBuffer.allocate(50);    // 将管道中的数据读取到缓冲Buffer中：    channel.read(buffer);    buffer.flip();    String msg = decoder.decode(buffer).toString();    System.out.println(＂收到：＂ + msg); }
　　将数据通过 Buffer 写入到 Channel 管道：  /**  * 发送消息  *  * @param msg  */ public void send(String msg) {     try {         SocketChannel client = (SocketChannel) clientKey.channel();         // 将数据通过 Buffer 写入到 Channel 管道：         client.write(encoder.encode(CharBuffer.wrap(msg)));     } catch (Exception e) {         e.printStackTrace();     } }
　　由此可知，管道中的数据传输、从管道中读取/写入数据，数据本身都是需要包装再 Buffer 缓冲中的  ！  原文：兴趣使然的草帽路飞
　　https://mp.weixin.qq.com/s/Sjn5pUUDUFb0VOFoeGnj1A