Java I/O體系的原理及應(yīng)用

這篇文章主要講解了“Java I/O體系的原理及應(yīng)用”，文中的講解內(nèi)容簡(jiǎn)單清晰，易于學(xué)習(xí)與理解，下面請(qǐng)大家跟著小編的思路慢慢深入，一起來(lái)研究和學(xué)習(xí)“Java I/O體系的原理及應(yīng)用”吧！

一、基礎(chǔ)概念

在介紹I/O原理之前，先重溫幾個(gè)基礎(chǔ)概念：

1. 操作系統(tǒng)與內(nèi)核

操作系統(tǒng)：管理計(jì)算機(jī)硬件與軟件資源的系統(tǒng)軟件

內(nèi)核：操作系統(tǒng)的核心軟件，負(fù)責(zé)管理系統(tǒng)的進(jìn)程、內(nèi)存、設(shè)備驅(qū)動(dòng)程序、文件和網(wǎng)絡(luò)系統(tǒng)等等，為應(yīng)用程序提供對(duì)計(jì)算機(jī)硬件的安全訪問(wèn)服務(wù)

2. 內(nèi)核空間和用戶空間

為了避免用戶進(jìn)程直接操作內(nèi)核，保證內(nèi)核安全，操作系統(tǒng)將內(nèi)存尋址空間劃分為兩部分：內(nèi)核空間(Kernel-space)，供內(nèi)核程序使用用戶空間(User-space)，供用戶進(jìn)程使用  為了安全，內(nèi)核空間和用戶空間是隔離的，即使用戶的程序崩潰了，內(nèi)核也不受影響。

3. 數(shù)據(jù)流

計(jì)算機(jī)中的數(shù)據(jù)是基于隨著時(shí)間變換高低電壓信號(hào)傳輸?shù)模@些數(shù)據(jù)信號(hào)連續(xù)不斷，有著固定的傳輸方向，類似水管中水的流動(dòng)，因此抽象數(shù)據(jù)流(I/O流)的概念：指一組有順序的、有起點(diǎn)和終點(diǎn)的字節(jié)集合，抽象出數(shù)據(jù)流的作用：實(shí)現(xiàn)程序邏輯與底層硬件解耦，通過(guò)引入數(shù)據(jù)流作為程序與硬件設(shè)備之間的抽象層，面向通用的數(shù)據(jù)流輸入輸出接口編程，而不是具體硬件特性，程序和底層硬件可以獨(dú)立靈活替換和擴(kuò)展。

二、I/O 工作原理

1. 磁盤I/O

典型I/O讀寫磁盤工作原理如下：

tips: DMA：全稱叫直接內(nèi)存存取(Direct Memory Access)，是一種允許外圍設(shè)備(硬件子系統(tǒng))直接訪問(wèn)系統(tǒng)主內(nèi)存的機(jī)制?；?DMA  訪問(wèn)方式，系統(tǒng)主內(nèi)存與硬件設(shè)備的數(shù)據(jù)傳輸可以省去CPU 的全程調(diào)度。

值得注意的是：

• 讀寫操作基于系統(tǒng)調(diào)用實(shí)現(xiàn)

• 讀寫操作經(jīng)過(guò)用戶緩沖區(qū)，內(nèi)核緩沖區(qū)，應(yīng)用進(jìn)程并不能直接操作磁盤

• 應(yīng)用進(jìn)程讀操作時(shí)需阻塞直到讀取到數(shù)據(jù)

2. 網(wǎng)絡(luò)I/O

這里先以最經(jīng)典的阻塞式I/O模型介紹：

tips:recvfrom，經(jīng)socket接收數(shù)據(jù)的函數(shù)

值得注意的是：

• 網(wǎng)絡(luò)I/O讀寫操作經(jīng)過(guò)用戶緩沖區(qū)，Sokcet緩沖區(qū)

• 服務(wù)端線程在從調(diào)用recvfrom開(kāi)始到它返回有數(shù)據(jù)報(bào)準(zhǔn)備好這段時(shí)間是阻塞的，recvfrom返回成功后，線程開(kāi)始處理數(shù)據(jù)報(bào)

三、Java I/O設(shè)計(jì)

1. I/O分類

Java中對(duì)數(shù)據(jù)流進(jìn)行具體化和實(shí)現(xiàn)，關(guān)于Java數(shù)據(jù)流一般關(guān)注以下幾個(gè)點(diǎn)：

• 流的方向從外部到程序，稱為輸入流;從程序到外部，稱為輸出流

• 流的數(shù)據(jù)單位程序以字節(jié)作為最小讀寫數(shù)據(jù)單元，稱為字節(jié)流，以字符作為最小讀寫數(shù)據(jù)單元，稱為字符流

• 流的功能角色

從/向一個(gè)特定的IO設(shè)備(如磁盤，網(wǎng)絡(luò))或者存儲(chǔ)對(duì)象(如內(nèi)存數(shù)組)讀/寫數(shù)據(jù)的流，稱為節(jié)點(diǎn)流;

對(duì)一個(gè)已有流進(jìn)行連接和封裝，通過(guò)封裝后的流來(lái)實(shí)現(xiàn)數(shù)據(jù)的讀/寫功能，稱為處理流(或稱為過(guò)濾流)。

2. I/O操作接口

java.io包下有一堆I/O操作類，初學(xué)時(shí)看了容易搞不懂，其實(shí)仔細(xì)觀察其中還是有規(guī)律：這些I/O操作類都是在繼承4個(gè)基本抽象流的基礎(chǔ)上，要么是節(jié)點(diǎn)流，要么是處理流。

(1) 四個(gè)基本抽象流

java.io包中包含了流式I/O所需要的所有類，java.io包中有四個(gè)基本抽象流，分別處理字節(jié)流和字符流：

• InputStream

• OutputStream

• Reader

• Writer

(2) 節(jié)點(diǎn)流

節(jié)點(diǎn)流I/O類名由節(jié)點(diǎn)流類型 + 抽象流類型組成，常見(jiàn)節(jié)點(diǎn)類型有：

• File文件

• Piped 進(jìn)程內(nèi)線程通信管道

• ByteArray / CharArray (字節(jié)數(shù)組 / 字符數(shù)組)

• StringBuffer / String (字符串緩沖區(qū) / 字符串)

節(jié)點(diǎn)流的創(chuàng)建通常是在構(gòu)造函數(shù)傳入數(shù)據(jù)源，例如：

FileReader reader = new FileReader(new File("file.txt")); FileWriter writer = new FileWriter(new File("file.txt"));

(3) 處理流

處理流I/O類名由對(duì)已有流封裝的功能 + 抽象流類型組成，常見(jiàn)功能有：

• 緩沖：對(duì)節(jié)點(diǎn)流讀寫的數(shù)據(jù)提供了緩沖的功能，數(shù)據(jù)可以基于緩沖批量讀寫，提高效率。常見(jiàn)有BufferedInputStream、BufferedOutputStream

• 字節(jié)流轉(zhuǎn)換為字符流：由InputStreamReader、OutputStreamWriter實(shí)現(xiàn)

• 字節(jié)流與基本類型數(shù)據(jù)相互轉(zhuǎn)換：這里基本數(shù)據(jù)類型數(shù)據(jù)如int、long、short，由DataInputStream、DataOutputStream實(shí)現(xiàn)

• 字節(jié)流與對(duì)象實(shí)例相互轉(zhuǎn)換：用于實(shí)現(xiàn)對(duì)象序列化，由ObjectInputStream、ObjectOutputStream實(shí)現(xiàn)

處理流的應(yīng)用了適配器/裝飾模式，轉(zhuǎn)換/擴(kuò)展已有流，處理流的創(chuàng)建通常是在構(gòu)造函數(shù)傳入已有的節(jié)點(diǎn)流或處理流：

FileOutputStream fileOutputStream = new FileOutputStream("file.txt"); // 擴(kuò)展提供緩沖寫 BufferedOutputStream bufferedOutputStream = new BufferedOutputStream(fileOutputStream);  // 擴(kuò)展提供提供基本數(shù)據(jù)類型寫 DataOutputStream out = new DataOutputStream(bufferedOutputStream);

3. Java NIO

(1) 標(biāo)準(zhǔn)I/O存在問(wèn)題

Java NIO(New I/O)是一個(gè)可以替代標(biāo)準(zhǔn)Java I/O API的IO API(從Java 1.4開(kāi)始)，Java  NIO提供了與標(biāo)準(zhǔn)I/O不同的I/O工作方式，目的是為了解決標(biāo)準(zhǔn) I/O存在的以下問(wèn)題：

A.  數(shù)據(jù)多次拷貝

標(biāo)準(zhǔn)I/O處理，完成一次完整的數(shù)據(jù)讀寫，至少需要從底層硬件讀到內(nèi)核空間，再讀到用戶文件，又從用戶空間寫入內(nèi)核空間，再寫入底層硬件。

此外，底層通過(guò)write、read等函數(shù)進(jìn)行I/O系統(tǒng)調(diào)用時(shí)，需要傳入數(shù)據(jù)所在緩沖區(qū)起始地址和長(zhǎng)度由于JVM  GC的存在，導(dǎo)致對(duì)象在堆中的位置往往會(huì)發(fā)生移動(dòng)，移動(dòng)后傳入系統(tǒng)函數(shù)的地址參數(shù)就不是真正的緩沖區(qū)地址了。

可能導(dǎo)致讀寫出錯(cuò)，為了解決上面的問(wèn)題，使用標(biāo)準(zhǔn)I/O進(jìn)行系統(tǒng)調(diào)用時(shí)，還會(huì)額外導(dǎo)致一次數(shù)據(jù)拷貝：把數(shù)據(jù)從JVM的堆內(nèi)拷貝到堆外的連續(xù)空間內(nèi)存(堆外內(nèi)存)。

所以總共經(jīng)歷6次數(shù)據(jù)拷貝，執(zhí)行效率較低。

B. 操作阻塞

傳統(tǒng)的網(wǎng)絡(luò)I/O處理中，由于請(qǐng)求建立連接(connect)，讀取網(wǎng)絡(luò)I/O數(shù)據(jù)(read)，發(fā)送數(shù)據(jù)(send)等操作是線程阻塞的。

// 等待連接 Socket socket = serverSocket.accept();  // 連接已建立，讀取請(qǐng)求消息 StringBuilder req = new StringBuilder(); byte[] recvByteBuf = new byte[1024]; int len; while ((len = socket.getInputStream().read(recvByteBuf)) != -1) {     req.append(new String(recvByteBuf, 0, len, StandardCharsets.UTF_8)); }  // 寫入返回消息 socket.getOutputStream().write(("server response msg".getBytes())); socket.shutdownOutput();

以上面服務(wù)端程序?yàn)槔?，?dāng)請(qǐng)求連接已建立，讀取請(qǐng)求消息，服務(wù)端調(diào)用read方法時(shí)，客戶端數(shù)據(jù)可能還沒(méi)就緒(例如客戶端數(shù)據(jù)還在寫入中或者傳輸中)，線程需要在read方法阻塞等待直到數(shù)據(jù)就緒。

為了實(shí)現(xiàn)服務(wù)端并發(fā)響應(yīng)，每個(gè)連接需要獨(dú)立的線程單獨(dú)處理，當(dāng)并發(fā)請(qǐng)求量大時(shí)為了維護(hù)連接，內(nèi)存、線程切換開(kāi)銷過(guò)大。

(2) Buffer

Java NIO核心三大核心組件是Buffer(緩沖區(qū))、Channel(通道)、Selector。

Buffer提供了常用于I/O操作的字節(jié)緩沖區(qū)，常見(jiàn)的緩存區(qū)有ByteBuffer, CharBuffer, DoubleBuffer,  FloatBuffer, IntBuffer, LongBuffer, ShortBuffer，分別對(duì)應(yīng)基本數(shù)據(jù)類型: byte, char, double,  float, int, long, short，下面介紹主要以最常用的ByteBuffer為例，Buffer底層支持Java堆外內(nèi)存和堆內(nèi)內(nèi)存。

堆外內(nèi)存是指與堆內(nèi)存相對(duì)應(yīng)的，把內(nèi)存對(duì)象分配在JVM堆以外的內(nèi)存，這些內(nèi)存直接受操作系統(tǒng)管理(而不是虛擬機(jī)，相比堆內(nèi)內(nèi)存，I/O操作中使用堆外內(nèi)存的優(yōu)勢(shì)在于：

• 不用被JVM GC線回收，減少GC線程資源占有

• 在I/O系統(tǒng)調(diào)用時(shí)，直接操作堆外內(nèi)存，可以節(jié)省一次堆外內(nèi)存和堆內(nèi)內(nèi)存的復(fù)制

ByteBuffer底層基于堆外內(nèi)存的分配和釋放基于malloc和free函數(shù)，對(duì)外allocateDirect方法可以申請(qǐng)分配堆外內(nèi)存，并返回繼承ByteBuffer類的DirectByteBuffer對(duì)象：

public static ByteBuffer allocateDirect(int capacity) {     return new DirectByteBuffer(capacity); }

堆外內(nèi)存的回收基于DirectByteBuffer的成員變量Cleaner類，提供clean方法可以用于主動(dòng)回收，Netty中大部分堆外內(nèi)存通過(guò)記錄定位Cleaner的存在，主動(dòng)調(diào)用clean方法來(lái)回收;另外，當(dāng)DirectByteBuffer對(duì)象被GC時(shí)，關(guān)聯(lián)的堆外內(nèi)存也會(huì)被回收。

tips：JVM參數(shù)不建議設(shè)置-XX:+DisableExplicitGC，因?yàn)椴糠忠蕾嘕ava  NIO的框架(例如Netty)在內(nèi)存異常耗盡時(shí)，會(huì)主動(dòng)調(diào)用System.gc()，觸發(fā)Full  GC，回收DirectByteBuffer對(duì)象，作為回收堆外內(nèi)存的最后保障機(jī)制，設(shè)置該參數(shù)之后會(huì)導(dǎo)致在該情況下堆外內(nèi)存得不到清理。

堆外內(nèi)存基于基礎(chǔ)ByteBuffer類的DirectByteBuffer類成員變量：Cleaner對(duì)象，這個(gè)Cleaner對(duì)象會(huì)在合適的時(shí)候執(zhí)行unsafe.freeMemory(address)，從而回收這塊堆外內(nèi)存。

Buffer可以見(jiàn)到理解為一組基本數(shù)據(jù)類型，存儲(chǔ)地址連續(xù)的的數(shù)組，支持讀寫操作，對(duì)應(yīng)讀模式和寫模式，通過(guò)幾個(gè)變量來(lái)保存這個(gè)數(shù)據(jù)的當(dāng)前位置狀態(tài)：capacity、  position、 limit：

• capacity 緩沖區(qū)數(shù)組的總長(zhǎng)度

• position 下一個(gè)要操作的數(shù)據(jù)元素的位置

• limit 緩沖區(qū)數(shù)組中不可操作的下一個(gè)元素的位置：limit <= capacity

(3) Channel

Channel(通道)的概念可以類比I/O流對(duì)象，NIO中I/O操作主要基于Channel：從Channel進(jìn)行數(shù)據(jù)讀取  ：創(chuàng)建一個(gè)緩沖區(qū)，然后請(qǐng)求Channel讀取數(shù)據(jù) 從Channel進(jìn)行數(shù)據(jù)寫入 ：創(chuàng)建一個(gè)緩沖區(qū)，填充數(shù)據(jù)，請(qǐng)求Channel寫入數(shù)據(jù)。

Channel和流非常相似，主要有以下幾點(diǎn)區(qū)別：

• Channel可以讀和寫，而標(biāo)準(zhǔn)I/O流是單向的

• Channel可以異步讀寫，標(biāo)準(zhǔn)I/O流需要線程阻塞等待直到讀寫操作完成

• Channel總是基于緩沖區(qū)Buffer讀寫

Java NIO中最重要的幾個(gè)Channel的實(shí)現(xiàn)：

• FileChannel：用于文件的數(shù)據(jù)讀寫，基于FileChannel提供的方法能減少讀寫文件數(shù)據(jù)拷貝次數(shù)，后面會(huì)介紹

• DatagramChannel：用于UDP的數(shù)據(jù)讀寫

• SocketChannel：用于TCP的數(shù)據(jù)讀寫，代表客戶端連接

• ServerSocketChannel：監(jiān)聽(tīng)TCP連接請(qǐng)求，每個(gè)請(qǐng)求會(huì)創(chuàng)建會(huì)一個(gè)SocketChannel，一般用于服務(wù)端

基于標(biāo)準(zhǔn)I/O中，我們第一步可能要像下面這樣獲取輸入流，按字節(jié)把磁盤上的數(shù)據(jù)讀取到程序中，再進(jìn)行下一步操作，而在NIO編程中，需要先獲取Channel，再進(jìn)行讀寫。

FileInputStream fileInputStream = new FileInputStream("test.txt"); FileChannel channel = fileInputStream.channel();

tips: FileChannel僅能運(yùn)行在阻塞模式下，文件異步處理的 I/O 是在JDK 1.7 才被加入的  java.nio.channels.AsynchronousFileChannel。

// server socket channel: ServerSocketChannel serverSocketChannel = ServerSocketChannel.open(); serverSocketChannel.bind(new InetSocketAddress(InetAddress.getLocalHost(), 9091));  while (true) {     SocketChannel socketChannel = serverSocketChannel.accept();     ByteBuffer buffer = ByteBuffer.allocateDirect(1024);     int readBytes = socketChannel.read(buffer);     if (readBytes > 0) {         // 從寫數(shù)據(jù)到buffer翻轉(zhuǎn)為從buffer讀數(shù)據(jù)         buffer.flip();         byte[] bytes = new byte[buffer.remaining()];         buffer.get(bytes);         String body = new String(bytes, StandardCharsets.UTF_8);         System.out.println("server 收到：" + body);     } }

(4) Selector

Selector(選擇器) ，它是Java NIO核心組件中的一個(gè)，用于檢查一個(gè)或多個(gè)NIO  Channel(通道)的狀態(tài)是否處于可讀、可寫。實(shí)現(xiàn)單線程管理多個(gè)Channel，也就是可以管理多個(gè)網(wǎng)絡(luò)連接。

Selector核心在于基于操作系統(tǒng)提供的I/O復(fù)用功能，單個(gè)線程可以同時(shí)監(jiān)視多個(gè)連接描述符，一旦某個(gè)連接就緒(一般是讀就緒或者寫就緒)，能夠通知程序進(jìn)行相應(yīng)的讀寫操作，常見(jiàn)有select、poll、epoll等不同實(shí)現(xiàn)。

Java NIO Selector基本工作原理如下：

• 初始化Selector對(duì)象，服務(wù)端ServerSocketChannel對(duì)象

• 向Selector注冊(cè)ServerSocketChannel的socket-accept事件

• 線程阻塞于selector.select()，當(dāng)有客戶端請(qǐng)求服務(wù)端，線程退出阻塞

• 基于selector獲取所有就緒事件，此時(shí)先獲取到socket-accept事件，向Selector注冊(cè)客戶端SocketChannel的數(shù)據(jù)就緒可讀事件事件

• 線程再次阻塞于selector.select()，當(dāng)有客戶端連接數(shù)據(jù)就緒，可讀

• 基于ByteBuffer讀取客戶端請(qǐng)求數(shù)據(jù)，然后寫入響應(yīng)數(shù)據(jù)，關(guān)閉channel

示例如下，完整可運(yùn)行代碼已經(jīng)上傳github(https://github.com/caison/caison-blog-demo)：

Selector selector = Selector.open(); ServerSocketChannel serverSocketChannel = ServerSocketChannel.open(); serverSocketChannel.bind(new InetSocketAddress(9091)); // 配置通道為非阻塞模式 serverSocketChannel.configureBlocking(false); // 注冊(cè)服務(wù)端的socket-accept事件 serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);  while (true) {     // selector.select()會(huì)一直阻塞，直到有channel相關(guān)操作就緒     selector.select();     // SelectionKey關(guān)聯(lián)的channel都有就緒事件     Iterator<SelectionKey> keyIterator = selector.selectedKeys().iterator();      while (keyIterator.hasNext()) {         SelectionKey key = keyIterator.next();         // 服務(wù)端socket-accept         if (key.isAcceptable()) {             // 獲取客戶端連接的channel             SocketChannel clientSocketChannel = serverSocketChannel.accept();             // 設(shè)置為非阻塞模式             clientSocketChannel.configureBlocking(false);             // 注冊(cè)監(jiān)聽(tīng)該客戶端channel可讀事件，并為channel關(guān)聯(lián)新分配的buffer             clientSocketChannel.register(selector, SelectionKey.OP_READ, ByteBuffer.allocateDirect(1024));         }          // channel可讀         if (key.isReadable()) {             SocketChannel socketChannel = (SocketChannel) key.channel();             ByteBuffer buf = (ByteBuffer) key.attachment();              int bytesRead;             StringBuilder reqMsg = new StringBuilder();             while ((bytesRead = socketChannel.read(buf)) > 0) {                 // 從buf寫模式切換為讀模式                 buf.flip();                 int bufbufRemain = buf.remaining();                 byte[] bytes = new byte[bufRemain];                 buf.get(bytes, 0, bytesRead);                 // 這里當(dāng)數(shù)據(jù)包大于byteBuffer長(zhǎng)度，有可能有粘包/拆包問(wèn)題                 reqMsg.append(new String(bytes, StandardCharsets.UTF_8));                 buf.clear();             }             System.out.println("服務(wù)端收到報(bào)文：" + reqMsg.toString());             if (bytesRead == -1) {                 byte[] bytes = "[這是服務(wù)回的報(bào)文的報(bào)文]".getBytes(StandardCharsets.UTF_8);                  int length;                 for (int offset = 0; offset < bytes.length; offset += length) {                     length = Math.min(buf.capacity(), bytes.length - offset);                     buf.clear();                     buf.put(bytes, offset, length);                     buf.flip();                     socketChannel.write(buf);                 }                 socketChannel.close();             }         }         // Selector不會(huì)自己從已selectedKeys中移除SelectionKey實(shí)例         // 必須在處理完通道時(shí)自己移除 下次該channel變成就緒時(shí)，Selector會(huì)再次將其放入selectedKeys中         keyIterator.remove();     } }

tips: Java NIO基于Selector實(shí)現(xiàn)高性能網(wǎng)絡(luò)I/O這塊使用起來(lái)比較繁瑣，使用不友好，一般業(yè)界使用基于Java  NIO進(jìn)行封裝優(yōu)化，擴(kuò)展豐富功能的Netty框架來(lái)優(yōu)雅實(shí)現(xiàn)。

四、高性能I/O優(yōu)化

下面結(jié)合業(yè)界熱門開(kāi)源項(xiàng)目介紹高性能I/O的優(yōu)化。

1. 零拷貝

零拷貝(zero  copy)技術(shù)，用于在數(shù)據(jù)讀寫中減少甚至完全避免不必要的CPU拷貝，減少內(nèi)存帶寬的占用，提高執(zhí)行效率，零拷貝有幾種不同的實(shí)現(xiàn)原理，下面介紹常見(jiàn)開(kāi)源項(xiàng)目中零拷貝實(shí)現(xiàn)。

(1) Kafka零拷貝

Kafka基于Linux 2.1內(nèi)核提供，并在2.4 內(nèi)核改進(jìn)的的sendfile函數(shù) + 硬件提供的DMA Gather  Copy實(shí)現(xiàn)零拷貝，將文件通過(guò)socket傳送。

函數(shù)通過(guò)一次系統(tǒng)調(diào)用完成了文件的傳送，減少了原來(lái)read/write方式的模式切換。同時(shí)減少了數(shù)據(jù)的copy, sendfile的詳細(xì)過(guò)程如下：

基本流程如下：

• 用戶進(jìn)程發(fā)起sendfile系統(tǒng)調(diào)用

• 內(nèi)核基于DMA Copy將文件數(shù)據(jù)從磁盤拷貝到內(nèi)核緩沖區(qū)

• 內(nèi)核將內(nèi)核緩沖區(qū)中的文件描述信息(文件描述符，數(shù)據(jù)長(zhǎng)度)拷貝到Socket緩沖區(qū)

• 內(nèi)核基于Socket緩沖區(qū)中的文件描述信息和DMA硬件提供的Gather Copy功能將內(nèi)核緩沖區(qū)數(shù)據(jù)復(fù)制到網(wǎng)卡

• 用戶進(jìn)程sendfile系統(tǒng)調(diào)用完成并返回

相比傳統(tǒng)的I/O方式，sendfile + DMA Gather  Copy方式實(shí)現(xiàn)的零拷貝，數(shù)據(jù)拷貝次數(shù)從4次降為2次，系統(tǒng)調(diào)用從2次降為1次，用戶進(jìn)程上下文切換次數(shù)從4次變成2次DMA Copy，大大提高處理效率。

Kafka底層基于java.nio包下的FileChannel的transferTo：

public abstract long transferTo(long position, long count, WritableByteChannel target)

transferTo將FileChannel關(guān)聯(lián)的文件發(fā)送到指定channel，當(dāng)Comsumer消費(fèi)數(shù)據(jù)，Kafka  Server基于FileChannel將文件中的消息數(shù)據(jù)發(fā)送到SocketChannel。

A. RocketMQ零拷貝

RocketMQ基于mmap + write的方式實(shí)現(xiàn)零拷貝：mmap()  可以將內(nèi)核中緩沖區(qū)的地址與用戶空間的緩沖區(qū)進(jìn)行映射，實(shí)現(xiàn)數(shù)據(jù)共享，省去了將數(shù)據(jù)從內(nèi)核緩沖區(qū)拷貝到用戶緩沖區(qū)。

tmp_buf = mmap(file, len); write(socket, tmp_buf, len);

mmap + write 實(shí)現(xiàn)零拷貝的基本流程如下：

• 用戶進(jìn)程向內(nèi)核發(fā)起系統(tǒng)mmap調(diào)用

• 將用戶進(jìn)程的內(nèi)核空間的讀緩沖區(qū)與用戶空間的緩存區(qū)進(jìn)行內(nèi)存地址映射

• 內(nèi)核基于DMA Copy將文件數(shù)據(jù)從磁盤復(fù)制到內(nèi)核緩沖區(qū)

• 用戶進(jìn)程mmap系統(tǒng)調(diào)用完成并返回

• 用戶進(jìn)程向內(nèi)核發(fā)起write系統(tǒng)調(diào)用

• 內(nèi)核基于CPU Copy將數(shù)據(jù)從內(nèi)核緩沖區(qū)拷貝到Socket緩沖區(qū)

• 內(nèi)核基于DMA Copy將數(shù)據(jù)從Socket緩沖區(qū)拷貝到網(wǎng)卡

• 用戶進(jìn)程write系統(tǒng)調(diào)用完成并返回

RocketMQ中消息基于mmap實(shí)現(xiàn)存儲(chǔ)和加載的邏輯寫在org.apache.rocketmq.store.MappedFile中，內(nèi)部實(shí)現(xiàn)基于nio提供的java.nio.MappedByteBuffer，基于FileChannel的map方法得到mmap的緩沖區(qū)：

// 初始化 this.fileChannel = new RandomAccessFile(this.file, "rw").getChannel(); thisthis.mappedByteBuffer = this.fileChannel.map(MapMode.READ_WRITE, 0, fileSize);

查詢CommitLog的消息時(shí)，基于mappedByteBuffer偏移量pos，數(shù)據(jù)大小size查詢：

public SelectMappedBufferResult selectMappedBuffer(int pos, int size) {     int readPosition = getReadPosition();     // ...各種安全校驗(yàn)          // 返回mappedByteBuffer視圖     ByteBuffer byteBuffer = this.mappedByteBuffer.slice();     byteBuffer.position(pos);     ByteBuffer byteBufferbyteBufferNew = byteBuffer.slice();     byteBufferNew.limit(size);     return new SelectMappedBufferResult(this.fileFromOffset + pos, byteBufferNew, size, this); }

tips: transientStorePoolEnable機(jī)制Java NIO  mmap的部分內(nèi)存并不是常駐內(nèi)存，可以被置換到交換內(nèi)存(虛擬內(nèi)存)，RocketMQ為了提高消息發(fā)送的性能，引入了內(nèi)存鎖定機(jī)制，即將最近需要操作的CommitLog文件映射到內(nèi)存，并提供內(nèi)存鎖定功能，確保這些文件始終存在內(nèi)存中，該機(jī)制的控制參數(shù)就是transientStorePoolEnable。

因此，MappedFile數(shù)據(jù)保存CommitLog刷盤有2種方式：

• 開(kāi)啟transientStorePoolEnable：寫入內(nèi)存字節(jié)緩沖區(qū)(writeBuffer) ->  從內(nèi)存字節(jié)緩沖區(qū)(writeBuffer)提交(commit)到文件通道(fileChannel) -> 文件通道(fileChannel) ->  flush到磁盤

• 未開(kāi)啟transientStorePoolEnable：寫入映射文件字節(jié)緩沖區(qū)(mappedByteBuffer) ->  映射文件字節(jié)緩沖區(qū)(mappedByteBuffer) -> flush到磁盤

RocketMQ 基于 mmap+write 實(shí)現(xiàn)零拷貝，適用于業(yè)務(wù)級(jí)消息這種小塊文件的數(shù)據(jù)持久化和傳輸 Kafka 基于 sendfile  這種零拷貝方式，適用于系統(tǒng)日志消息這種高吞吐量的大塊文件的數(shù)據(jù)持久化和傳輸。

tips: Kafka 的索引文件使用的是 mmap+write 方式，數(shù)據(jù)文件發(fā)送網(wǎng)絡(luò)使用的是 sendfile 方式。

B. Netty零拷貝

Netty 的零拷貝分為兩種：

• 基于操作系統(tǒng)實(shí)現(xiàn)的零拷貝，底層基于FileChannel的transferTo方法

• 基于Java 層操作優(yōu)化，對(duì)數(shù)組緩存對(duì)象(ByteBuf )進(jìn)行封裝優(yōu)化，通過(guò)對(duì)ByteBuf數(shù)據(jù)建立數(shù)據(jù)視圖，支持ByteBuf  對(duì)象合并，切分，當(dāng)?shù)讓觾H保留一份數(shù)據(jù)存儲(chǔ)，減少不必要拷貝

2. 多路復(fù)用

Netty中對(duì)Java NIO功能封裝優(yōu)化之后，實(shí)現(xiàn)I/O多路復(fù)用代碼優(yōu)雅了很多：

// 創(chuàng)建mainReactor NioEventLoopGroup boosGroup = new NioEventLoopGroup(); // 創(chuàng)建工作線程組 NioEventLoopGroup workerGroup = new NioEventLoopGroup();  final ServerBootstrap serverBootstrap = new ServerBootstrap(); serverBootstrap      // 組裝NioEventLoopGroup     .group(boosGroup, workerGroup)      // 設(shè)置channel類型為NIO類型     .channel(NioServerSocketChannel.class)     // 設(shè)置連接配置參數(shù)     .option(ChannelOption.SO_BACKLOG, 1024)     .childOption(ChannelOption.SO_KEEPALIVE, true)     .childOption(ChannelOption.TCP_NODELAY, true)     // 配置入站、出站事件handler     .childHandler(new ChannelInitializer<NioSocketChannel>() {         @Override         protected void initChannel(NioSocketChannel ch) {             // 配置入站、出站事件channel             ch.pipeline().addLast(...);             ch.pipeline().addLast(...);         }     });  // 綁定端口 int port = 8080; serverBootstrap.bind(port).addListener(future -> {     if (future.isSuccess()) {         System.out.println(new Date() + ": 端口[" + port + "]綁定成功!");     } else {         System.err.println("端口[" + port + "]綁定失敗!");     } });

3. 頁(yè)緩存(PageCache)

頁(yè)緩存(PageCache)是操作系統(tǒng)對(duì)文件的緩存，用來(lái)減少對(duì)磁盤的 I/O  操作，以頁(yè)為單位的，內(nèi)容就是磁盤上的物理塊，頁(yè)緩存能幫助程序?qū)ξ募M(jìn)行順序讀寫的速度幾乎接近于內(nèi)存的讀寫速度，主要原因就是由于OS使用PageCache機(jī)制對(duì)讀寫訪問(wèn)操作進(jìn)行了性能優(yōu)化：

頁(yè)緩存讀取策略：當(dāng)進(jìn)程發(fā)起一個(gè)讀操作 (比如，進(jìn)程發(fā)起一個(gè) read() 系統(tǒng)調(diào)用)，它首先會(huì)檢查需要的數(shù)據(jù)是否在頁(yè)緩存中：

• 如果在，則放棄訪問(wèn)磁盤，而直接從頁(yè)緩存中讀取

• 如果不在，則內(nèi)核調(diào)度塊 I/O 操作從磁盤去讀取數(shù)據(jù)，并讀入緊隨其后的少數(shù)幾個(gè)頁(yè)面(不少于一個(gè)頁(yè)面，通常是三個(gè)頁(yè)面)，然后將數(shù)據(jù)放入頁(yè)緩存中

頁(yè)緩存寫策略：當(dāng)進(jìn)程發(fā)起write系統(tǒng)調(diào)用寫數(shù)據(jù)到文件中，先寫到頁(yè)緩存，然后方法返回。此時(shí)數(shù)據(jù)還沒(méi)有真正的保存到文件中去，Linux  僅僅將頁(yè)緩存中的這一頁(yè)數(shù)據(jù)標(biāo)記為“臟”，并且被加入到臟頁(yè)鏈表中。

然后，由flusher  回寫線程周期性將臟頁(yè)鏈表中的頁(yè)寫到磁盤，讓磁盤中的數(shù)據(jù)和內(nèi)存中保持一致，最后清理“臟”標(biāo)識(shí)。在以下三種情況下，臟頁(yè)會(huì)被寫回磁盤：

• 空閑內(nèi)存低于一個(gè)特定閾值

• 臟頁(yè)在內(nèi)存中駐留超過(guò)一個(gè)特定的閾值時(shí)

• 當(dāng)用戶進(jìn)程調(diào)用 sync() 和 fsync() 系統(tǒng)調(diào)用時(shí)

RocketMQ中，ConsumeQueue邏輯消費(fèi)隊(duì)列存儲(chǔ)的數(shù)據(jù)較少，并且是順序讀取，在page cache機(jī)制的預(yù)讀取作用下，Consume  Queue文件的讀性能幾乎接近讀內(nèi)存，即使在有消息堆積情況下也不會(huì)影響性能，提供了2種消息刷盤策略：

• 同步刷盤：在消息真正持久化至磁盤后RocketMQ的Broker端才會(huì)真正返回給Producer端一個(gè)成功的ACK響應(yīng)

• 異步刷盤，能充分利用操作系統(tǒng)的PageCache的優(yōu)勢(shì)，只要消息寫入PageCache即可將成功的ACK返回給Producer端。消息刷盤采用后臺(tái)異步線程提交的方式進(jìn)行，降低了讀寫延遲，提高了MQ的性能和吞吐量

Kafka實(shí)現(xiàn)消息高性能讀寫也利用了頁(yè)緩存，這里不再展開(kāi)。

感謝各位的閱讀，以上就是“Java I/O體系的原理及應(yīng)用”的內(nèi)容了，經(jīng)過(guò)本文的學(xué)習(xí)后，相信大家對(duì)Java I/O體系的原理及應(yīng)用這一問(wèn)題有了更深刻的體會(huì)，具體使用情況還需要大家實(shí)踐驗(yàn)證。這里是億速云，小編將為大家推送更多相關(guān)知識(shí)點(diǎn)的文章，歡迎關(guān)注！