原生 JDK 网络编程- NIO（六）

本是古典何须时尚 2022-10-29 07:37 137阅读 0赞

今天我们分享原生 JDK 网络编程- NIO

**1、什么是**  **NIO** **？**

NIO  库是在  JDK 1.4  中引入的。 NIO  弥补了原来的  I/O  的不足，它在标准  Java  代码中提供了高速的、面向块的 I/O 。 NIO  翻译成  no-blocking io  或者  new io  都说得通。

**2、NIO*和 BIO***  ***的主要区别***

面向流与面向缓冲

Java NIO  和  IO  之间第一个最大的区别是，IO 是面向流的， NIO  是面向缓冲区的。 Java IO 面向流意味着每次从流中读一个或多个字节，直至读取所有字节，它们没有被缓存在任何地 方。此外，它不能前后移动流中的数据。如果需要前后移动从流中读取的数据，需要先将它缓存到一个缓冲区。 Java NIO  的缓冲导向方法略有不同。数据读取到一个它稍后处理的缓冲区，需要时可在缓冲区中前后移动。这就增加了处理过程中的灵活性。但是，还需要检查是否该缓冲区中包含所有需要处理的数据。而且，需确保当更多的数据读入缓冲区时，不要覆盖缓冲区里尚未处理的数据。

阻塞与非阻塞  IO

Java IO  的各种流是阻塞的。这意味着，当一个线程调用  read()  或  write() 时，该线程被 阻塞，直到有一些数据被读取，或数据完全写入。该线程在此期间不能再干任何事情了。 Java NIO 的非阻塞模式，使一个线程从某通道发送请求读取数据，但是它仅能得到目 前可用的数据，如果目前没有数据可用时，就什么都不会获取。而不是保持线程阻塞，所以 直至数据变的可以读取之前，该线程可以继续做其他的事情。 非阻塞写也是如此。一个线 程请求写入一些数据到某通道，但不需要等待它完全写入，这个线程同时可以去做别的事情。 线程通常将非阻塞 IO  的空闲时间用于在其它通道上执行  IO  操作，所以一个单独的线程现在可以管理多个输入和输出通道（channel ）。

**3、NIO**  **三大核心组件**

NIO  有三大核心组件： Selector  选择器、 Channel  管道、 buffer  缓冲区。

***Selector***

Selector  的英文含义是 “ 选择器 ” ，也可以称为为 “ 轮询代理器 ” 、 “ 事件订阅器 ” 、 “channel 容器管理机” 都行。 Java NIO 的选择器允许一个单独的线程来监视多个输入通道，你可以注册多个通道使用 一个选择器(Selectors) ，然后使用一个单独的线程来操作这个选择器，进而“选择”通道： 这些通道里已经有可以处理的输入，或者选择已准备写入的通道。这种选择机制，使得一个 单独的线程很容易来管理多个通道。 应用程序将向 Selector  对象注册需要它关注的  Channel ，以及具体的某一个  Channel  会 对哪些 IO  事件感兴趣。 Selector  中也会维护一个 “ 已经注册的  Channel” 的容器。

***Channels***

通道，被建立的一个应用程序和操作系统交互事件、传递内容的渠道（注意是连接到操 作系统）。那么既然是和操作系统进行内容的传递，那么说明应用程序可以通过通道读取数 据，也可以通过通道向操作系统写数据，而且可以同时进行读写。

所有被  Selector （选择器）注册的通道，只能是继承了  SelectableChannel  类的子类。

ServerSocketChannel ：应用服务器程序的监听通道。只有通过这个通道，应用程序才能向操作系统注册支持“ 多路复用  IO” 的端口监听。同时支持  UDP  协议和  TCP  协议。

ScoketChannel ： TCP Socket  套接字的监听通道，一个  Socket  套接字对应了一个客户端 IP ：端口 到 服务器  IP ：端口的通信连接。 通道中的数据总是要先读到一个 Buffer ，或者总是要从一个  Buffer  中写入。

***buffer***  ***缓冲区***

后面会详细讲到其中的  api  等相关内容。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70][]

**4、重要概念**  **SelectionKey**

***什么是***  ***SelectionKey***

SelectionKey 是一个抽象类 , 表示 selectableChannel 在 Selector 中注册的标识 . 每个 Channel 向 Selector  注册时 , 都将会创建一个  SelectionKey 。 SelectionKey  将  Channel  与  Selector  建立了 关系, 并维护了  channel  事件。 可以通过 cancel  方法取消键 , 取消的键不会立即从  selector  中移除 , 而是添加到 cancelledKeys 中 , 在下一次  select  操作时移除它 . 所以在调用某个  key  时 , 需要使用  isValid  进行 校验.

***SelectionKey***  ***类型和就绪条件***

在向  Selector  对象注册感兴趣的事件时，JAVA NIO 共定义了四种： OP\_READ 、 OP\_WRITE 、 OP\_CONNECT、 OP\_ACCEPT （定义在  SelectionKey  中），分别对应读、写、请求连接、接受连接等网络 Socket  操作。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1][]

***服务端和客户端分别感兴趣的类型***

ServerSocketChannel  和  SocketChannel  可以注册自己感兴趣的操作类型，当对应操作类 型的就绪条件满足时 OS  会通知  channel ，下表描述各种  Channel  允许注册的操作类型， Y  表 示允许注册，N  表示不允许注册，其中服务器  SocketChannel  指由服务器 ServerSocketChannel.accept()返回的对象。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 2][]

服务器启动  ServerSocketChannel ，关注  OP\_ACCEPT  事件，

客户端启动  SocketChannel ，连接服务器，关注  OP\_CONNECT  事件

服务器接受连接，启动一个服务器的  SocketChannel ，这个  SocketChannel  可以关注

OP\_READ 、 OP\_WRITE  事件，一般连接建立后会直接关注  OP\_READ  事件

客户端这边的客户端  SocketChannel  发现连接建立后，可以关注  OP\_READ 、 OP\_WRITE

事件，一般是需要客户端需要发送数据了才关注  OP\_READ  事件

连接建立后客户端与服务器端开始相互发送消息（读写），根据实际情况来关注 OP\_READ 、 OP\_WRITE 事件。

**5、原生**  **JDK**  **网络编程** **- Buffer**

Buffer  用于和  NIO  通道进行交互。数据是从通道读入缓冲区，从缓冲区写入到通道中的。 以写为例，应用程序都是将数据写入缓冲，再通过通道把缓冲的数据发送出去，读也是一样， 数据总是先从通道读到缓冲，应用程序再读缓冲的数据。 缓冲区本质上是一块可以写入数据，然后可以从中读取数据的内存（ 其实就是数组）。 这块内存被包装成 NIO Buffer  对象，并提供了一组方法，用来方便的访问该块内存。

***重要属性***

capacity

作为一个内存块， Buffer  有一个固定的大小值，也叫 “capacity”. 你只能往里写  capacity 个 byte 、 long ， char  等类型。一旦  Buffer  满了，需要将其清空（通过读数据或者清除数据） 才能继续写数据往里写数据。

position

当你写数据到  Buffer  中时， position  表示当前能写的位置。初始的  position  值为  0. 当一 个 byte 、 long  等数据写到  Buffer  后，  position  会向前移动到下一个可插入数据的  Buffer  单 元。position  最大可为  capacity – 1. 当读取数据时，也是从某个特定位置读。当将 Buffer  从写模式切换到读模式， position 会被重置为 0.  当从  Buffer  的  position  处读取数据时，position 向前移动到下一个可读的位置。

limit

在写模式下， Buffer  的  limit  表示你最多能往  Buffer  里写多少数据。 写模式下， limit  等 于 Buffer  的  capacity 。 当切换 Buffer  到读模式时，  limit  表示你最多能读到多少数据。因此，当切换  Buffer  到 读模式时，limit  会被设置成写模式下的  position  值。换句话说，你能读到之前写入的所有数据（limit  被设置成已写数据的数量，这个值在写模式下就是  position ）

***Buffer***  ***的分配***

要想获得一个  Buffer  对象首先要进行分配。 每一个  Buffer  类都有  **allocate**  方法 ( 可以在 堆上分配，也可以在直接内存上分配) 。

分配  48  字节  capacity  的  ByteBuffer  的例子 :ByteBuffer buf = ByteBuffer.allocate(48);

分配一个可存储  1024  个字符的  CharBuffer ： CharBuffer buf = CharBuffer.allocate(1024);

**wrap**  **方法** ：把一个  byte  数组或  byte  数组的一部分包装成  ByteBuffer ：

ByteBuffer wrap(byte [] array)
    ByteBuffer wrap(byte [] array, int offset, int length)

***直接内存***

HeapByteBuffer  与  DirectByteBuffer ，在原理上，前者可以看出分配的  buffer  是在  heap 区域的，其实真正 flush  到远程的时候会先拷贝到直接内存，再做下一步操作；在  NIO  的框 架下，很多框架会采用 DirectByteBuffer  来操作，这样分配的内存不再是在  java heap  上，经 过性能测试，可以得到非常快速的网络交互，在大量的网络交互下，一般速度会比 HeapByteBuffer 要快速好几倍。 直接内存（Direct Memory ）并不是虚拟机运行时数据区的一部分，也不是  Java  虚拟机 规范中定义的内存区域，但是这部分内存也被频繁地使用，而且也可能导致 OutOfMemoryError 异常出现。 NIO 可以使用  Native  函数库直接分配堆外内存，然后通过一个存储在  Java  堆里面的 DirectByteBuffer 对象作为这块内存的引用进行操作。这样能在一些场景中显著提高性能，因为避免了在 Java  堆和  Native  堆中来回复制数据。

测试代码如下：

public class AllocateBuffer {
        public static void main(String[] args) {
    
            OperatingSystemMXBean osmxb = (OperatingSystemMXBean)
                    ManagementFactory.getOperatingSystemMXBean();
    
    
            System.out.println("----------Test allocate--------");
            System.out.println("before allocate:"
                    + osmxb.getFreePhysicalMemorySize());
    
            /*堆上分配*/
            ByteBuffer buffer = ByteBuffer.allocate(200000);
            System.out.println("buffer = " + buffer);
            System.out.println("after allocate:"
                    + osmxb.getFreePhysicalMemorySize());
    
            /* 这部分用的直接内存*/
            ByteBuffer directBuffer = ByteBuffer.allocateDirect(200000);
            System.out.println("directBuffer = " + directBuffer);
            System.out.println("after direct allocate:"
                    + osmxb.getFreePhysicalMemorySize());
    
            System.out.println("----------Test wrap--------");
            byte[] bytes = new byte[32];
            buffer = ByteBuffer.wrap(bytes);
            System.out.println(buffer);
    
            buffer = ByteBuffer.wrap(bytes, 10, 10);
            System.out.println(buffer);
        }
    }

直接内存（堆外内存）与堆内存比较

直接内存申请空间耗费更高的性能，当频繁申请到一定量时尤为明显

直接内存  IO  读写的性能要优于普通的堆内存，在多次读写操作的情况下差异明显

***Buffer***  ***的读写***

向  Buffer  中写数据

**写数据到**  **Buffer**  **有两种方式：**

**读取**  **Channel**  **写到**  **Buffer** **。**

**通过**  **Buffer**  **的**  **put()** **方法写到**  **Buffer**  **里。**

从  Channel  写到  Buffer  的例子

**int**  bytesRead = inChannel.read(buf); //read into buffer.

通过  put  方法写  Buffer  的例子：

buf.put(127);

put  方法有很多版本，允许你以不同的方式把数据写入到  Buffer  中。例如， 写到一个 指定的位置，或者把一个字节数组写入到 Buffer 。在比如：

put(byte b) 相对写，向 position  的位置写入一个  byte ，并将  postion+1 ，为下次读写 作准备。

flip() 方法

flip  方法将  Buffer  从写模式切换到读模式。调用  flip() 方法会将  position  设回  0 ，并将  limit

设置成之前  position  的值。 换句话说，position  现在用于标记读的位置， limit  表示之前写进了多少个  byte 、 char  等

——  现在能读取多少个  byte 、 char  等。

从  Buffer  中读取数据

**从**  **Buffer**  **中读取数据有两种方式：**

**1.**  **从**  **Buffer**  **读取数据写入到**  **Channel** **。**

**2.**  **使用**  **get()** **方法从**  **Buffer**  **中读取数据。**

从  Buffer  读取数据到  Channel  的例子：

**int**  bytesWritten = inChannel.write(buf);

使用  get() 方法从  Buffer  中读取数据的例子

**byte**  aByte = buf.get();

get  方法有很多版本，允许你以不同的方式从  Buffer  中读取数据。例如，从指定  position 读取，或者从 Buffer  中读取数据到字节数组，再比如 get()属于相对读，从  position  位置读取一个  byte ，并将  position+1 ，为下次读写作准备 ;

使用  Buffer  读写数据常见步骤

1.  写入数据到  Buffer

2.  调用  flip() 方法

3.  从  Buffer  中读取数据

4.  调用  clear() 方法或者  compact() 方法，准备下一次的写入

当向  buffer  写入数据时， buffer  会记录下写了多少数据。一旦要读取数据，需要通过  flip() 方法将 Buffer  从写模式切换到读模式。在读模式下，可以读取之前写入到  buffer  的所有数据。 一旦读完了所有的数据，就需要清空缓冲区，让它可以再次被写入。有两种方式能清空 缓冲区：调用 clear() 或  compact() 方法。 clear() 方法会清空整个缓冲区。 compact() 方法只会清除已经读过的数据。

其他常用操作

**绝对读写**

put(int index, byte b)

绝对写，向  byteBuffer  底层的  bytes  中下标为  index  的位置插入 byte b，不改变  position  的值。

get(int index) 属于绝对读，读取  byteBuffer  底层的  bytes  中下标为  index  的  byte ，不改变 position 。

更多  Buffer  实现的细节参考  JavaDoc 。

**rewind()** **方法**

Buffer.rewind() 将  position  设回  0 ，所以你可以重读  Buffer  中的所有数据。limit 保持不变， 仍然表示能从 Buffer  中读取多少个元素（ byte 、 char  等）。

**clear()** **与**  **compact()** **方法**

一旦读完 Buffer 中的数据，需要让 Buffer 准备好再次被写入。可以通过 clear() 或 compact() 方法来完成。 如果调用的是 clear() 方法， position  将被设回  0 ， limit  被设置成  capacity  的值。换句话 说，Buffer  被清空了。 Buffer  中的数据并未清除，只是这些标记告诉我们可以从哪里开始往 Buffer 里写数据。 如果 Buffer  中有一些未读的数据，调用  clear() 方法，数据将 “ 被遗忘 ” ，意味着不再有任 何标记会告诉你哪些数据被读过，哪些还没有。如果 Buffer 中仍有未读的数据，且后续还需要这些数据，但是此时想要先先写些数据， 那么使用 compact() 方法。 compact()方法将所有未读的数据拷贝到  Buffer  起始处。然后将  position  设到最后一个未 读元素正后面。limit  属性依然像  clear() 方法一样，设置成  capacity 。现在  Buffer  准备好写数 据了，但是不会覆盖未读的数据。

**mark()** **与**  **reset()** **方法**

通过调用  Buffer.mark() 方法，可以标记  Buffer  中的一个特定  position 。之后可以通过调 用 Buffer.reset() 方法恢复到这个  position 。例如：

buffer.mark();//call buffer.get() a couple of times, e.g. during parsing.

buffer.reset(); //set position back to mark.

**equals()** **与**  **compareTo()** **方法**

可以使用  equals() 和  compareTo() 方法两个  Buffer 。

**equals()**

当满足下列条件时，表示两个  Buffer  相等：

1.  有相同的类型（ byte 、 char 、 int  等）。

2. Buffer  中剩余的  byte 、 char  等的个数相等。

3. Buffer  中所有剩余的  byte 、 char  等都相同。

如你所见， equals  只是比较  Buffer  的一部分，不是每一个在它里面的元素都比较。实际 上，它只比较 Buffer  中的剩余元素。

**compareTo()** **方法**

compareTo() 方法比较两个  Buffer  的剩余元素 (byte 、 char  等 ) ， 如果满足下列条件，则 认为一个 Buffer“ 小于 ” 另一个  Buffer ：

1.  第一个不相等的元素小于另一个  Buffer  中对应的元素 。

2.  所有元素都相等，但第一个  Buffer  比另一个先耗尽 ( 第一个  Buffer  的元素个数比另一 个少) 。

测试代码如下：

public class BufferMethod {
        public static void main(String[] args) {
    
            System.out.println("------Test get and put-------------");
            ByteBuffer buffer = ByteBuffer.allocate(32);
            buffer.put((byte) 'a')//0
                    .put((byte) 'b')//1
                    .put((byte) 'c')//2
                    .put((byte) 'd')//3
                    .put((byte) 'e')//4
                    .put((byte) 'f');//5
            System.out.println("before flip()" + buffer);
            /* 转换为读取模式*/
            buffer.flip();
            System.out.println("before get():" + buffer);
            System.out.println((char)buffer.get());
            System.out.println("after get():" + buffer);
    
            /* position移动两位*/
            byte[] dst = new byte[10];
            buffer.get(dst, 0, 2);
            System.out.println("after get(dst, 0, 2):" + buffer);
            System.out.println("dst:" + new String(dst));
    
            /*绝对读写*/
            System.out.println("--------Test 绝对读写-------");
            ByteBuffer bb = ByteBuffer.allocate(32);
            System.out.println("before put(byte):" + bb);
            System.out.println("after put(byte):" + bb.put((byte) 'z'));
            /* put(2,(byte) 'c')不改变position的位置*/
            bb.put(2, (byte) 'c');
            System.out.println("after put(2,(byte) 'c'):" + bb);
            System.out.println(new String(bb.array()));
    
            /* get(index)不影响position的值*/
            System.out.println((char) buffer.get(2));
            System.out.println("after get(index):" + buffer);
    
            System.out.println("--------Test Clear And Compact--------");
            ByteBuffer buffer2 = ByteBuffer.allocate(32);
            buffer2.put((byte) 'x');
            System.out.println("before clear:" + buffer2);
            buffer2.clear();
            System.out.println("after clear:" + buffer2);
            System.out.println(new String(buffer2.array()));
            /*放入4个字节，position移动到下个可写入的位置，也就是4*/
            buffer2.put("abcd".getBytes());
            System.out.println("before compact:" + buffer2);
            buffer2.flip();//将position设回0，并将limit设置成之前position的值*/
            System.out.println("after flip:" + buffer2);
            /*compact()方法将所有未读的数据拷贝到Buffer起始处。*/
            /* 然后将position设到最后一个未读元素正后面。*/
            System.out.println("还有数据未读，个数:" + buffer2.remaining());
            buffer2.compact();
            System.out.println("after compact:" + buffer2);
            System.out.println(new String(buffer2.array()));
    
            System.out.println("--------Test rewind--------");
            buffer.clear();
            buffer.position(10);/*移动position到10*/
            buffer.limit(15);/*限定最大可写入的位置为15*/
            System.out.println("before rewind:" + buffer);
            buffer.rewind();/*将position设回0*/
            System.out.println("before rewind:" + buffer);
    
            System.out.println("--------Test mark AND reset----------");
            buffer = ByteBuffer.allocate(20);
            System.out.println("buffer = " + buffer);
            buffer.clear();
            buffer.position(5);/*移动position到5*/
            buffer.mark();/*记录当前position的位置*/
            buffer.position(10);/*移动position到10*/
            System.out.println("before reset:" + buffer);
            buffer.reset();/*复位position到记录的地址*/
            System.out.println("after reset:" + buffer);
    
        }
    }

***Buffer***  ***方法总结***

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 3][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 4][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 5][]

**原生**  **JDK**  **网络编程** **-NIO**  **实战**

1、服务端代码:

public class NioServer {
        private static NioServerHandle nioServerHandle;
    
        public static void main(String[] args){
            nioServerHandle = new NioServerHandle(DEFAULT_PORT);
            new Thread(nioServerHandle,"Server").start();
        }
    
    }

NioServerHandle核心类：

public class NioServerHandle implements Runnable{
    
        private volatile boolean started;
        private ServerSocketChannel serverSocketChannel;
        private Selector selector;
    
    
    
        /**
         * 构造方法
         * @param port 指定要监听的端口号
         */
        public NioServerHandle(int port) {
            try {
                /*创建选择器的实例*/
                selector = Selector.open();
                /*创建ServerSocketChannel的实例*/
                serverSocketChannel = ServerSocketChannel.open();
    
                /*设置通道为非阻塞模式*/
                serverSocketChannel.configureBlocking(false);
                /*绑定端口*/
                serverSocketChannel.socket().bind(new InetSocketAddress(port));
                /*注册事件，表示关心客户端连接*/
                serverSocketChannel.register(selector,SelectionKey.OP_ACCEPT);
    
                started = true;
                System.out.println("服务器已启动，端口号："+port);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    
        @Override
        public void run() {
            while(started){
                try {
                    /*获取当前有哪些事件*/
                    selector.select(1000);
                    /*获取事件的集合*/
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    Iterator<SelectionKey> iterator = selectionKeys.iterator();
                    while(iterator.hasNext()){
                        SelectionKey key = iterator.next();
                        /*我们必须首先将处理过的 SelectionKey 从选定的键集合中删除。
                        如果我们没有删除处理过的键，那么它仍然会在主集合中以一个激活
                        的键出现，这会导致我们尝试再次处理它。*/
                        iterator.remove();
                        handleInput(key);
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    
        /*处理事件的发生*/
        private void handleInput(SelectionKey key) throws IOException {
            if(key.isValid()){
                /*处理新接入的客户端的请求*/
                if(key.isAcceptable()){
                    /*获取关心当前事件的Channel*/
                    ServerSocketChannel ssc
                            = (ServerSocketChannel) key.channel();
                    /*接受连接*/
                    SocketChannel sc = ssc.accept();
                    System.out.println("==========建立连接=========");
                    sc.configureBlocking(false);
                    /*关注读事件*/
                    sc.register(selector,SelectionKey.OP_READ);
                }
                /*处理对端的发送的数据*/
                if(key.isReadable()){
                    SocketChannel sc = (SocketChannel) key.channel();
                    /*创建ByteBuffer，开辟一个缓冲区*/
                    ByteBuffer buffer = ByteBuffer.allocate(1024);
                    /*从通道里读取数据，然后写入buffer*/
                    int readBytes = sc.read(buffer);
                    if(readBytes>0){
                        /*将缓冲区当前的limit设置为position,position=0，
                        用于后续对缓冲区的读取操作*/
                        buffer.flip();
                        /*根据缓冲区可读字节数创建字节数组*/
                        byte[] bytes = new byte[buffer.remaining()];
                        /*将缓冲区可读字节数组复制到新建的数组中*/
                        buffer.get(bytes);
                        String message = new String(bytes,"UTF-8");
                        System.out.println("服务器收到消息："+message);
                        /*处理数据*/
                        String result = Const.response(message);
                        /*发送应答消息*/
                        doWrite(sc,result);
    
                    }else if(readBytes<0){
                        /*取消特定的注册关系*/
                        key.cancel();
                        /*关闭通道*/
                        sc.close();
                    }
                }
            }
        }
    
        /*发送应答消息*/
        private void doWrite(SocketChannel sc,String response) throws IOException {
            byte[] bytes = response.getBytes();
            ByteBuffer buffer = ByteBuffer.allocate(bytes.length);
            buffer.put(bytes);
            buffer.flip();
            sc.write(buffer);
        }
    
    
        public void stop(){
            started = false;
        }
    
    }

2、客户端代码：

public class NioClient {
        private static NioClientHandle nioClientHandle;
    
        public static void start(){
            nioClientHandle = new NioClientHandle(DEFAULT_SERVER_IP,DEFAULT_PORT);
           // nioClientHandle = new NioClientHandle(DEFAULT_SERVER_IP,9999);
            new Thread(nioClientHandle,"Server").start();
        }
        //向服务器发送消息
        public static boolean sendMsg(String msg) throws Exception{
            nioClientHandle.sendMsg(msg);
            return true;
        }
        public static void main(String[] args) throws Exception {
            start();
            Scanner scanner = new Scanner(System.in);
            while(NioClient.sendMsg(scanner.next()));
        }
    }

核心代码：

public class NioClientHandle implements Runnable{
        private String host;
        private int port;
        private volatile boolean started;
        private Selector selector;
        private SocketChannel socketChannel;
    
        public NioClientHandle(String ip, int port) {
            this.host = ip;
            this.port = port;
    
            try {
                /*创建选择器的实例*/
                selector = Selector.open();
                /*创建ServerSocketChannel的实例*/
                socketChannel = SocketChannel.open();
                /*设置通道为非阻塞模式*/
                socketChannel.configureBlocking(false);
    
                started = true;
            } catch (IOException e) {
                e.printStackTrace();
            }
    
    
        }
        public void stop(){
            started = false;
        }
        @Override
        public void run() {
            try{
                doConnect();
            }catch(IOException e){
                e.printStackTrace();
                System.exit(1);
            }
    
            //循环遍历selector
            while(started){
                try{
                    //无论是否有读写事件发生，selector每隔1s被唤醒一次
                    selector.select(1000);
                    //获取当前有哪些事件可以使用
                    Set<SelectionKey> keys = selector.selectedKeys();
                    //转换为迭代器
                    Iterator<SelectionKey> it = keys.iterator();
                    SelectionKey key = null;
                    while(it.hasNext()){
                        key = it.next();
                        /*我们必须首先将处理过的 SelectionKey 从选定的键集合中删除。
                        如果我们没有删除处理过的键，那么它仍然会在主集合中以一个激活
                        的键出现，这会导致我们尝试再次处理它。*/
                        it.remove();
                        try{
                            handleInput(key);
                        }catch(Exception e){
                            if(key != null){
                                key.cancel();
                                if(key.channel() != null){
                                    key.channel().close();
                                }
                            }
                        }
                    }
                }catch(Exception e){
                    e.printStackTrace();
                    System.exit(1);
                }
            }
            //selector关闭后会自动释放里面管理的资源
            if(selector != null)
                try{
                    selector.close();
                }catch (Exception e) {
                    e.printStackTrace();
                }
        }
    
        //具体的事件处理方法
        private void handleInput(SelectionKey key) throws IOException{
            if(key.isValid()){
                //获得关心当前事件的channel
                SocketChannel sc = (SocketChannel) key.channel();
                //连接事件
                if(key.isConnectable()){
                    if(sc.finishConnect()){
                        socketChannel.register(selector,
                            SelectionKey.OP_READ);}
                    else System.exit(1);
                }
                //有数据可读事件
                if(key.isReadable()){
                    //创建ByteBuffer，并开辟一个1M的缓冲区
                    ByteBuffer buffer = ByteBuffer.allocate(1024);
                    //读取请求码流，返回读取到的字节数
                    int readBytes = sc.read(buffer);
                    //读取到字节，对字节进行编解码
                    if(readBytes>0){
                        //将缓冲区当前的limit设置为position,position=0，
                        // 用于后续对缓冲区的读取操作
                        buffer.flip();
                        //根据缓冲区可读字节数创建字节数组
                        byte[] bytes = new byte[buffer.remaining()];
                        //将缓冲区可读字节数组复制到新建的数组中
                        buffer.get(bytes);
                        String result = new String(bytes,"UTF-8");
                        System.out.println("客户端收到消息：" + result);
                    }
                    //链路已经关闭，释放资源
                    else if(readBytes<0){
                        key.cancel();
                        sc.close();
                    }
                }
            }
        }
    
        private void doWrite(SocketChannel channel,String request)
                throws IOException {
            //将消息编码为字节数组
            byte[] bytes = request.getBytes();
            //根据数组容量创建ByteBuffer
            ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
            //将字节数组复制到缓冲区
            writeBuffer.put(bytes);
            //flip操作
            writeBuffer.flip();
            //发送缓冲区的字节数组
            /*关心事件和读写网络并不冲突*/
            channel.write(writeBuffer);
        }
    
        private void doConnect() throws IOException{
            /*非阻塞的连接*/
            if(socketChannel.connect(new InetSocketAddress(host,port))){
                socketChannel.register(selector,SelectionKey.OP_READ);
            }else{
                socketChannel.register(selector,SelectionKey.OP_CONNECT);
            }
        }
    
        //写数据对外暴露的API
        public void sendMsg(String msg) throws Exception{
            doWrite(socketChannel, msg);
        }
    }

工具类：

public class Const {
        public static int DEFAULT_PORT = 12345;
        public static String DEFAULT_SERVER_IP = "127.0.0.1";
    
        /*根据输入信息拼接出一个应答信息*/
        public static String response(String msg){
            return "Hello,"+msg+",Now is "+new java.util.Date(
                    System.currentTimeMillis()).toString() ;
        }
    }

3、测试执行：

客户端：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 6][]

服务端：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 7][]

4、简要分析：

1)、实例化 Selector

Selector 对象是通过调用静态工厂方法 open()来实例化的，如下：

Selector Selector=Selector.open()；

2)、设置非阻塞

要实现 Selector 管理 Channel，需要将 channel 注册到相应的 Selector 上，如下：

channel.configureBlocking(false);

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

通过调用通道的 register()方法会将它注册到一个选择器上。与 Selector 一起使用时， Channel 必须处于非阻塞模式下，否则将抛出 IllegalBlockingModeException 异常，这意 味着不能将 FileChannel 与 Selector 一起使用，因为 FileChannel 不能切换到非阻塞模式， 而套接字通道都可以。另外通道一旦被注册，将不能再回到阻塞状态，此时若调用通道的 configureBlocking(true)将抛出 BlockingModeException 异常。

择器对通道的 read 和 write 操作感兴趣，那么选择器在检查该通道时，只会检查通道的 read 和 write 操作是否已经处在就绪状态。  具体的操作类型和通道上能被支持的操作类型前面已经讲述过。 如果 Selector 对通道的多操作类型感兴趣，可以用“位或”操作符来实现：  int interestSet=SelectionKey.OP\_READ|SelectionKey.OP\_WRITE;

同时 一个 Channel 仅仅可以被注册到一个 Selector 一次, 如果将 Channel 注册 到 Selector 多次, 那么其实就是相当于更新 SelectionKey 的 interest set。

通过 SelectionKey 可以判断 Selector 是否对 Channel 的某种事件感兴趣，比如 int interestSet = selectionKey.interestOps();

boolean isInterestedInAccept = (interestSet & SelectionKey.OP\_ACCEPT) == SelectionKey.OP\_ACCEPT；

通过 SelctionKey 对象的 readyOps()来获取相关通道已经就绪的操作。它是 interest 集合的子集，并且表示了 interest 集合中从上次调用 select()以后已经就绪的那些操作。

JAVA 中定义几个方法用来检查这些操作是否就绪，比如 selectionKey.isAcceptable(); 同时，通过 SelectionKey 可以取出这个 SelectionKey 所关联的 Selector 和 Channel。

如果我们要取消关联关系，怎么办？SelectionKey 对象的 cancel()方法来取消特定的 注册关系。  在实际的应用中，我们还可以为 SelectionKey 绑定附加对象，在需要的时候取出。

SelectionKey key=channel.register(selector,SelectionKey.OP\_READ,theObject); 或 selectionKey.attach(theObject);

取出这个附加对象，通过： Object attachedObj = key.attachment();

3)、选择通道类型

在实际运行中，我们通过 Selector 的 select（）方法可以选择已经准备就绪的通道（这些通道包含你感兴趣的的事件）。

下面是 Selector 几个重载的 select()方法：

select():阻塞到至少有一个通道在你注册的事件上就绪了。

select(long timeout)：和 select()一样，但最长阻塞事件为 timeout 毫秒。

selectNow():非阻塞，立刻返回。

select()方法返回的 int 值表示有多少通道已经就绪,是自上次调用 select()方法后有 多少通道变成就绪状态。

一旦调用 select()方法，并且返回值不为 0 时，则可以通过调用 Selector 的 selectedKeys()方法来访问已选择键集合。

Set selectedKeys=selector.selectedKeys(); 这个时候，循环遍历 selectedKeys 集中的每个键，并检测各个键所对应的通道的就绪 事件，再通过 SelectionKey 关联的 Selector 和 Channel 进行实际的业务处理。

注意每次迭代末尾的 keyIterator.remove()调用。Selector 不会自己从已选择键集中 移除 SelectionKey 实例。必须在处理完通道时自己移除，否则的话，下次该通道变成就绪

时，Selector 会再次将其放入已选择键集中。

原生 JDK 网络编程- NIO分享完毕，下篇我们分享NIO的 Reactor 模式，敬请期待！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70]: /images/20221024/23dd427eff5340cd8118cbb73b23a1b7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1]: /images/20221024/6f7a788415f4470c8888dc4bc81533a3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 2]: /images/20221024/2b3b9d1f84ba462b886f8d67d17c8a36.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 3]: /images/20221024/e3801cb6ce764fbda5bcc73abbcdfd74.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 4]: /images/20221024/c703f9cf7a41457d87b96d96c968f831.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 5]: /images/20221024/eee04bfe1cc74924b9be80a465b5fb72.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 6]: /images/20221024/1d5a784dfe5f47d0b8cfc5b72ab82748.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 7]: /images/20221024/6efe0d01bac2437a92f51d9b24da8f03.png