阅读一个分布式框架必备的NIO知识有哪些

阅读一个分布式框架必备的NIO知识有哪些

这篇文章主要介绍“阅读一个分布式框架必备的NIO知识有哪些”,在日常操作中,相信很多人在阅读一个分布式框架必备的NIO知识有哪些问题上存在疑惑,小编查阅了各式资料,整理出简单好用的操作方法,希望对大家解答”阅读一个分布式框架必备的NIO知识有哪些”的疑惑有所帮助!接下来,请跟着小编一起来学习吧!

阅读一个分布式框架必备的NIO知识有哪些

 一、开篇

阅读一个分布式开源项目的时候,最重要的就是了解这个项目的通信框架。

因为一个分布式的开源框架,通常是集群部署的,不同的节点和节点之间需要相互通信来完成复杂的功能,而阅读到这些源码的时候,如果不了解它通信机制的话,就会迷失在代码里,像走进了一片原始森林。

比如 HDFS ,使用的通信框架是自己封装的 Hadoop Rpc;Spark 底层通信就是用的 Netty;而最近阅读的 Kafka  源码,底层使用的是原生的 Java NIO。

所以本次,我们来聊一聊 Java NIO 的那些主要的知识点。

二、多图弄懂 NIO 三大核心概念

谈到 NIO,就会有三个核心的概念:通道、缓冲、选择器。

直接开门见山,或许听起来会有点迷茫,我们需要从头开始说。

1、通道

以前在并发要求不是很高的情况下,是 CPU 来全权处理输入输出的(中断),如下图:

阅读一个分布式框架必备的NIO知识有哪些

用户程序向服务端发起读写请求,cpu 直接处理这些请求。这样有一个弊端,当 IO 请求非常多的时候,会大量占用  CPU,使得整个系统的处理能力会下降。

随着计算机的发展,出现了一种新的方式,使用 DMA 来全权处理 IO 请求,如下图:

阅读一个分布式框架必备的NIO知识有哪些

DMA 是 Direct Memory Access,直接内存访问控制。

为什么要增加这个设备呢?是因为 CPU  中断方式不能满足数据传输速度的要求,因为在中断方式下,每次中断需要保存断点和现场,中断返回时,要恢复断点和现场。

所有这些原因,使得中断方式难以满足高速外设对传输速度的要求。

所以,就有了 DMA 这样的设备,在 DMA 方式的数据传输过程中,当 I/O 设备需要进行数据传送时,通过 DMA 控制器向 CPU 提出 DMA  传送请求,CPU 响应之后将让出系统总线,由 DMA 控制器接管总线进行数据传输,而此时 CPU 除了做一些初始化操作之外,可以去做自己的事情。

但是有了 DMA,仍然满足不了业务快速发展的需要,因为当 I/O 请求过多时,会出现总线冲突的问题。

所以后面就出现了通道(Channel),它和 DMA 不同的地方是,通道有自己的指令系统和程序,是一个协处理器;而 DMA  只能实现固定的数据传送控制。

阅读一个分布式框架必备的NIO知识有哪些

而 Java NIO 中的 Channel ,就是对上图中通道的实现。

2、缓冲

理解了通道的概念,缓冲区也很好理解了。

通道表示打开到 I/O 设备的(例如:文件、套接字)的连接,但是通道本身并不存储数据。真正作为数据传输载体的是缓冲区。

当应用程序要写数据时,需要先把数据写到缓冲区里,然后由通道负责把缓冲区的数据发送到目的地(文件、磁盘、网络),然后再从缓冲区把数据取出来。

若需要使用 NIO 系统,需要获取用于连接 I/O 设备的通道以及用于容纳数据的缓冲区,然后操作缓冲区,对数据进行处理。

阅读一个分布式框架必备的NIO知识有哪些

3、选择器

选择器也叫做多路复用器,是一种非阻塞式的 I/O 。既然谈到了非阻塞式,必然要先谈谈阻塞式。阻塞式如下图所示:

阅读一个分布式框架必备的NIO知识有哪些

客户端向服务端发出一个读写请求时,服务端的线程会一直看内核地址空间是否有数据了。

客户端没有数据发送过来时,服务端的线程会一直等待,在此期间是什么事情都做不了的。

直到客户端有数据发送过来,会把数据从内核地址空间拷贝到用户地址空间,然后才读取到了数据的。

这就导致如果有大量的请求过来,后面的请求要等待前面的请求执行完毕,会造成大量的排队,无法充分利用 cpu 资源,性能就会急剧下降。

再看看选择器是如何工作的。

阅读一个分布式框架必备的NIO知识有哪些

现在客户端服务端之间通信是用通道+缓冲区的,那么所有的通道都会注册到选择器上来。选择器会监控这些通道的 I/O 状态,比如连接、读、写的情况。

当某一个通道上的某个事件完全就绪时,选择器才会把这个任务分配到服务端的一个或者多个线程上。

当客户端没有事件准备好时,服务端的线程是不会阻塞的,它可以做自己的事情,直到客户端事件就绪,才会去处理。

这种非阻塞式相比较阻塞式,可以进一步的利用 cpu 资源。

三、理解了概念,再来学 API

1、缓冲区的 API

要彻底理解缓冲区,必须知道缓冲区的四个属性,mark,position,limit,capacity,只需要跑一遍代码就知道了。

(1)分配一定大小的缓冲区

//1.分配一个指定大小的缓冲区 ByteBuffer buffer = ByteBuffer.allocate(10); System.out.println("---------alocate"); System.out.println("position:" + buffer.position()); System.out.println("limit:" + buffer.limit()); System.out.println("capacity:" + buffer.capacity());

运行结果:

---------alocate----------- position:0 limit:10 capacity:10

阅读一个分布式框架必备的NIO知识有哪些

这里我们分配了 10 个字节的缓冲区,也就是在 ByteBuffer 的 final byte[] hb; 属性上开辟了 10 个字节的空间。

所以容量 capacity 为 10 , limit 可读写数据的最大位置 也是 10 ,position 为可以操作数据的位置为 0 。

(2)往缓冲区写数据

// 2.写入数据到缓冲区 String str = "abcde"; System.out.println("------------put------------"); buffer.put(str.getBytes(StandardCharsets.UTF_8)); System.out.println("position:" + buffer.position()); System.out.println("limit:" + buffer.limit()); System.out.println("capacity:" + buffer.capacity());

运行结果:

------------put------------ position:5 limit:10 capacity:10

阅读一个分布式框架必备的NIO知识有哪些

这里我们往缓冲区写了 5 个字节的数据,那么 capacity 和 limit 都还是10,但是 position 为 5 了,因为前面已经写入了 5  个了

(3)切换成读数据的模式

// 3.切换成读数据的模式 buffer.flip(); System.out.println("------------flip------------"); System.out.println("position:" + buffer.position()); System.out.println("limit:" + buffer.limit()); System.out.println("capacity:" + buffer.capacity());

那我们现在想从缓冲区读取一些数据出来,就需要切换成 flip 模式,flip 会改变一些属性的值

运行结果:

------------flip------------ position:0 limit:5 capacity:10

阅读一个分布式框架必备的NIO知识有哪些

flip 会改变 position 的值为 0 ,并且 limit 为5,表示我要从头开始读,并且只能读到 5 的位置

(4)读取一些数据

// 4. 读取数据 System.out.println("------------get------------"); byte[] dest = new byte[buffer.limit()]; buffer.get(dest); System.out.println(new String(dest,0,dest.length)); System.out.println("position:" + buffer.position()); System.out.println("limit:" + buffer.limit()); System.out.println("capacity:" + buffer.capacity());

运行结果:

------------get------------ abcde position:5 limit:5 capacity:10

阅读一个分布式框架必备的NIO知识有哪些

读取了数据之后,position 就变成 5 了,表示我已经读取到 5 了。

(5)重复读

//5.rewind() buffer.rewind(); System.out.println("------------rewind------------"); System.out.println("position:" + buffer.position()); System.out.println("limit:" + buffer.limit()); System.out.println("capacity:" + buffer.capacity());

运行结果:

------------rewind------------ position:0 limit:5 capacity:10

阅读一个分布式框架必备的NIO知识有哪些

rewind 表示重复读取 buffer 里面的数据

(6)清除数据

//6.clear() buffer.clear(); System.out.println("------------clear------------"); System.out.println("position:" + buffer.position()); System.out.println("limit:" + buffer.limit()); System.out.println("capacity:" + buffer.capacity());

运行结果:

------------clear------------ position:0 limit:10 capacity:10

阅读一个分布式框架必备的NIO知识有哪些

clear() 之后,position 回到了 0 ,limit 回到了 10,又可以重头开始写数据了,能写 10 个字节。

但是要注意的是,缓冲里面的数据并没有清空掉,数据还在里面,处于被“遗忘”状态。这几个指针回到了最初的状态。

(7)标记

这是第四个属性:mark。

mark 可以记录 position 的位置。可以通过 reset() 方法回到 mark 的位置。

 @Test     public void test2() {         // 分配 10 个字节         String str = "abcde";         ByteBuffer buffer = ByteBuffer.allocate(10);         buffer.put(str.getBytes(StandardCharsets.UTF_8));          // 切换到读模式,读取 2 个字节         buffer.flip();         byte[] dest = new byte[buffer.limit()];         buffer.get(dest, 0, 2);         System.out.println(new String(dest, 0, 2));         System.out.println(buffer.position());          // mark 一下记录当前位置         buffer.mark();          // 又读取两个字节         buffer.get(dest, 2, 2);         System.out.println(new String(dest, 2, 2));         System.out.println(buffer.position());          // reset,回到 mark 的位置         buffer.reset();         System.out.println(buffer.position());     }  执行结果:  ```tex ab 2 cd 4 2

2、使用通道、缓冲区、选择器完成一个网络程序

(1)服务端

@Test  public void testServer() throws IOException {      ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();      serverSocketChannel.configureBlocking(false);       serverSocketChannel.bind(new InetSocketAddress(8989));       Selector selector = Selector.open();      serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);       while (selector.select() > 0) {          Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();          while (iterator.hasNext()) {              SelectionKey key = iterator.next();              if (key.isAcceptable()) {                  SocketChannel socketChannel = serverSocketChannel.accept();                  socketChannel.configureBlocking(false);                  socketChannel.register(selector, SelectionKey.OP_READ);              } else if (key.isReadable()) {                  SocketChannel channel = (SocketChannel) key.channel();                  ByteBuffer byteBuffer = ByteBuffer.allocate(1024);                  int len = 0;                  while ((len = channel.read(byteBuffer)) > 0) {                      byteBuffer.flip();                      System.out.println(new String(byteBuffer.array(), 0, len));                      byteBuffer.clear();                  }              }          }           iterator.remove();      }  }

1、首先使用 ServerSocketChannel.open(),打开一个通道,设置成非阻塞模式;

2、绑定到 8989 端口上;

3、把通道注册到选择器上;

4、while 循环,选择器上是否有事件,如果事件是客户端的连接事件,则打开一个  SocketChannel,注册成非阻塞模式,并且往选择器上注册一个读数据的事件;

5、当客户端发送数据过来的时候,就可以打开一个通道,读取缓冲区上的数据;

6、并且此时,服务端是可以同时接受多个客户端的请求的。

(2)客户端

@Test  public void testClient() throws IOException {      SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress("127.0.0.1", 8989));      socketChannel.configureBlocking(false);       ByteBuffer byteBuffer = ByteBuffer.allocate(1024);      byteBuffer.put(new Date().toString().getBytes(StandardCharsets.UTF_8));      byteBuffer.flip();      socketChannel.write(byteBuffer);      byteBuffer.clear();       socketChannel.close();   }

1、客户端打开一个 SocketChannel,配置成非阻塞模式;

2、使用 ByteBuffer 发送数据(注意发送之前,要 flip);

3、关闭通道。

到此,关于“阅读一个分布式框架必备的NIO知识有哪些”的学习就结束了,希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习,快去试试吧!若想继续学习更多相关知识,请继续关注网站,小编会继续努力为大家带来更多实用的文章!