快捷導(dǎo)航

Java同步非阻塞模式NIO處理IO數(shù)據(jù)

更新時(shí)間：2023年10月01日 11:08:07 作者：goyeer

這篇文章主要介紹了Java同步非阻塞模式NIO處理IO數(shù)據(jù),服務(wù)器實(shí)現(xiàn)模式為一個(gè)請(qǐng)求一個(gè)線程,即客戶端發(fā)送的鏈接請(qǐng)求都會(huì)注冊(cè)到選擇器上,選擇器輪詢到連接有IO請(qǐng)求時(shí)才啟動(dòng)一個(gè)線程進(jìn)行處理,需要的朋友可以參考下

一、概述

NIO（Non-Blocking IO）是同步非阻塞方式來(lái)處理IO數(shù)據(jù)。服務(wù)器實(shí)現(xiàn)模式為一個(gè)請(qǐng)求一個(gè)線程，即客戶端發(fā)送的鏈接請(qǐng)求都會(huì)注冊(cè)到選擇器上，選擇器輪詢到連接有IO請(qǐng)求時(shí)才啟動(dòng)一個(gè)線程進(jìn)行處理。

二、常用概念

同步(synchronous)：調(diào)用方式指應(yīng)用（Application）,調(diào)用方發(fā)起有一個(gè)功能調(diào)用時(shí)，在沒(méi)有得到功能的結(jié)果之前，該調(diào)用不會(huì)返回。也就是說(shuō)調(diào)用方會(huì)一直等待被調(diào)用方返回功能的結(jié)果。
異步(asynchronous)：調(diào)用方發(fā)起一個(gè)功能調(diào)用時(shí)，沒(méi)有得到功能的結(jié)果立即返回，后續(xù)被調(diào)用方再通過(guò)回調(diào)等手段，把功能的結(jié)構(gòu)通知調(diào)用方。也就是調(diào)用方立即得到返回，但是返回中不包含執(zhí)行的結(jié)果。

同步和異步強(qiáng)調(diào)的是消息通信機(jī)制 (synchronous communication/ asynchronous communication)。所謂同步，就是在發(fā)出一個(gè)"調(diào)用"時(shí)，在沒(méi)有得到結(jié)果之前，該“調(diào)用”就不返回。但是一旦調(diào)用返回，就得到返回值了。換句話說(shuō)，就是由“調(diào)用者”主動(dòng)等待這個(gè)“調(diào)用”的結(jié)果。而異步則是相反，"調(diào)用"在發(fā)出之后，這個(gè)調(diào)用就直接返回了，所以沒(méi)有返回結(jié)果。換句話說(shuō)，當(dāng)一個(gè)異步過(guò)程調(diào)用發(fā)出后，調(diào)用者不會(huì)立刻得到結(jié)果。而是在"調(diào)用"發(fā)出后，"被調(diào)用者"通過(guò)狀態(tài)、通知來(lái)通知調(diào)用者，或通過(guò)回調(diào)函數(shù)處理這個(gè)調(diào)用

阻塞：線程發(fā)起一個(gè)調(diào)用時(shí), 在調(diào)用返回之前, 線程會(huì)被阻塞, 在這個(gè)狀態(tài)下會(huì)交出當(dāng)前CPU的使用權(quán)而暫停；也就是調(diào)用方會(huì)等待調(diào)用結(jié)果, 調(diào)用阻塞了調(diào)用方的線程, 線程不在運(yùn)行處理中。
非阻塞：線程發(fā)起一個(gè)調(diào)用時(shí), 調(diào)用會(huì)立即返回, 避免線程被阻塞。但是, 返回的結(jié)果只是被調(diào)用方當(dāng)前狀態(tài)的值, 實(shí)際使用時(shí), 調(diào)用方需要輪詢, 直到返回結(jié)果符合預(yù)期(直到數(shù)據(jù)準(zhǔn)備好)。

阻塞和非阻塞強(qiáng)調(diào)的是程序在等待調(diào)用結(jié)果（消息，返回值）時(shí)的狀態(tài). 阻塞調(diào)用是指調(diào)用結(jié)果返回之前，當(dāng)前線程會(huì)被掛起。調(diào)用線程只有在得到結(jié)果之后才會(huì)返回。非阻塞調(diào)用指在不能立刻得到結(jié)果之前，該調(diào)用不會(huì)阻塞當(dāng)前線程。對(duì)于同步調(diào)用來(lái)說(shuō)，很多時(shí)候當(dāng)前線程還是激活的狀態(tài)，只是從邏輯上當(dāng)前函數(shù)沒(méi)有返回而已，即同步等待時(shí)什么都不干，白白占用著資源。

同步阻塞 IO[BIO - BlockingIO]：在此種方式下，用戶進(jìn)程在發(fā)起一個(gè) IO 操作以后，必須等待 IO 操作的完成，只有當(dāng)真正完成了 IO 操作以后，用戶進(jìn)程才能運(yùn)行。 JAVA傳統(tǒng)的 IO 模型屬于此種方式。
同步非阻塞 IO[Non-Blocking IO]：在此種方式下，用戶進(jìn)程發(fā)起一個(gè) IO 操作以后邊可返回做其它事情，但是用戶進(jìn)程需要時(shí)不時(shí)的詢問(wèn) IO 操作是否就緒，這就要求用戶進(jìn)程不停的去詢問(wèn)，從而引入不必要的 CPU 資源浪費(fèi)。其中目前 JAVA 的 NIO 就屬于同步非阻塞 IO 。
異步阻塞 IO[IO Multiplexing]：此種方式下是指應(yīng)用發(fā)起一個(gè) IO 操作以后，不等待內(nèi)核 IO 操作的完成，等內(nèi)核完成 IO 操作以后會(huì)通知應(yīng)用程序，這其實(shí)就是同步和異步最關(guān)鍵的區(qū)別，同步必須等待或者主動(dòng)的去詢問(wèn) IO 是否完成，那么為什么說(shuō)是阻塞的呢？因?yàn)榇藭r(shí)是通過(guò) select 系統(tǒng)調(diào)用來(lái)完成的，而 select 函數(shù)本身的實(shí)現(xiàn)方式是阻塞的，而采用 select 函數(shù)有個(gè)好處就是它可以同時(shí)監(jiān)聽(tīng)多個(gè)文件句柄，從而提高系統(tǒng)的并發(fā)性！
異步非阻塞 IO[Asynchronous IO]: 在此種模式下，用戶進(jìn)程只需要發(fā)起一個(gè) IO 操作然后立即返回，等 IO 操作真正的完成以后，應(yīng)用程序會(huì)得到 IO 操作完成的通知，此時(shí)用戶進(jìn)程只需要對(duì)數(shù)據(jù)進(jìn)行處理就好了，不需要進(jìn)行實(shí)際的 IO 讀寫(xiě)操作，因?yàn)?真正的 IO讀取或者寫(xiě)入操作已經(jīng)由內(nèi)核完成了。目前 Java 中還沒(méi)有支持此種 IO 模型。

三、NIO的實(shí)現(xiàn)原理

Java的NIO主要由三個(gè)核心部分組成：Channel(通道)、Buffer(緩沖區(qū))、Selector。

所有的IO在NIO中都從一個(gè)Channel開(kāi)始，數(shù)據(jù)可以從Channel讀到Buffer中，也可以從Buffer寫(xiě)到Channel中。Channel有好幾種類(lèi)型，其中比較常用的有FileChannel、DatagramChannel、SocketChannel、ServerSocketChannel等，這些通道涵蓋了UDP和TCP網(wǎng)絡(luò)IO以及文件IO。

Buffer本質(zhì)上是一塊可以寫(xiě)入數(shù)據(jù)，然后可以從中讀取數(shù)據(jù)的內(nèi)存。這塊內(nèi)存被包裝成NIO Buffer對(duì)象，并提供了一組方法，用來(lái)方便的訪問(wèn)該塊內(nèi)存。Java NIO里關(guān)鍵的Buffer實(shí)現(xiàn)有CharBuffer、ByteBuffer、ShortBuffer、IntBuffer、LongBuffer、FloatBuffer、DoubleBuffer。這些Buffer覆蓋了你能通過(guò)IO發(fā)送的基本數(shù)據(jù)類(lèi)型，即byte、short、int、long、float、double、char。

Buffer對(duì)象包含三個(gè)重要的屬性，分別是capacity、position、limit，其中position和limit的含義取決于Buffer處在讀模式還是寫(xiě)模式。但不管Buffer處在什么模式，capacity的含義總是一樣的。

capacity：作為一個(gè)內(nèi)存塊，Buffer有個(gè)固定的最大值，就是capacity。Buffer只能寫(xiě)capacity個(gè)數(shù)據(jù)，一旦Buffer滿了，需要將其清空才能繼續(xù)寫(xiě)數(shù)據(jù)往里寫(xiě)數(shù)據(jù)。

position：當(dāng)寫(xiě)數(shù)據(jù)到Buffer中時(shí)，position表示當(dāng)前的位置。初始的position值為0。當(dāng)一個(gè)數(shù)據(jù)寫(xiě)到Buffer后， position會(huì)向前移動(dòng)到下一個(gè)可插入數(shù)據(jù)的Buffer單元。position最大可為capacity–1。當(dāng)讀取數(shù)據(jù)時(shí)，也是從某個(gè)特定位置讀。當(dāng)將Buffer從寫(xiě)模式切換到讀模式，position會(huì)被重置為0。當(dāng)從Buffer的position處讀取數(shù)據(jù)時(shí)，position向前移動(dòng)到下一個(gè)可讀的位置。

limit：在寫(xiě)模式下，Buffer的limit表示最多能往Buffer里寫(xiě)多少數(shù)據(jù)，此時(shí)limit等于capacity。當(dāng)切換Buffer到讀模式時(shí)， limit表示你最多能讀到多少數(shù)據(jù)，此時(shí)limit會(huì)被設(shè)置成寫(xiě)模式下的position值。

Selector允許單線程處理多個(gè) Channel，如果你的應(yīng)用打開(kāi)了多個(gè)連接（通道），但每個(gè)連接的流量都很低，使用Selector就會(huì)很方便。要使用Selector，得向Selector注冊(cè)Channel，然后調(diào)用它的select()方法。這個(gè)方法會(huì)一直阻塞到某個(gè)注冊(cè)的通道有事件就緒。一旦這個(gè)方法返回，線程就可以處理這些事件，事件例如有新連接進(jìn)來(lái)，數(shù)據(jù)接收等。

四、NIO代碼實(shí)現(xiàn)

客戶端實(shí)現(xiàn)

步驟

創(chuàng)建SocketChannel 通道
切換異步非阻塞模式configureBlocking(false)
設(shè)置緩沖區(qū)大小ByteBuffer.allocate(1024)
值寫(xiě)入緩沖區(qū) buffer.put(input.getBytes())
緩沖區(qū)中的值寫(xiě)入通道中channel.write()

代碼演示

public static void main(String[] args) throws IOException {
      //創(chuàng)建通道
      SocketChannel channel=SocketChannel.open(new InetSocketAddress("127.0.0.1",6001));
      //切換異步非阻塞模式
      channel.configureBlocking(false);
      //設(shè)置緩沖去大小
      ByteBuffer buffer=ByteBuffer.allocate(1024);
      System.out.println("輸入傳輸值：");
      //獲取鍵盤(pán)輸入的值
      Scanner scanner = new Scanner(System.in);
      while (scanner.hasNext()){
          String input=scanner.next();
          //把獲取的值寫(xiě)入緩沖區(qū)中
          buffer.put(input.getBytes());
          buffer.flip();
          //把緩沖區(qū)中的值寫(xiě)入通道中
          channel.write(buffer);
          buffer.clear();
       }
       channel.close();
 }

服務(wù)端實(shí)現(xiàn)

步驟

創(chuàng)建ServerSocketChannel通道
切換異步非阻塞模式configureBlocking(false)
綁定連接
獲取選擇器 Selector open = Selector.open()
將通道注冊(cè)到選擇器，并指定監(jiān)聽(tīng)接受事件
輪訓(xùn)式獲取選擇已準(zhǔn)備就緒的事件
獲取當(dāng)前選擇器所有注冊(cè)的監(jiān)聽(tīng)事件
獲取準(zhǔn)備就緒的事件
判斷是什么事件準(zhǔn)備就緒
接受就緒，獲取客戶端連接
設(shè)置非阻塞異步模式
將通道注冊(cè)到服務(wù)器上

代碼演示

public static void main(String[] args) throws IOException {
    //創(chuàng)建通道
    ServerSocketChannel channel=ServerSocketChannel.open();
    //切換到異步非阻塞模式
    channel.configureBlocking(false);
    //綁定鏈接
    channel.bind(new InetSocketAddress(6001));
        //獲取選擇器
        Selector open = Selector.open();
        //將通道注冊(cè)到選擇器,并指定監(jiān)聽(tīng)接受事件
        channel.register(open, SelectionKey.OP_ACCEPT);
        //輪訓(xùn)式獲取選擇已經(jīng)準(zhǔn)備就緒的事件
        while(open.select() > 0) {
            //獲取當(dāng)前選擇器所有注冊(cè)的監(jiān)聽(tīng)事件
            Iterator<SelectionKey> it = open.selectedKeys().iterator();
            while(it.hasNext()) {
                //獲取準(zhǔn)備就緒的事件
                SelectionKey sk = it.next();
                //判斷是什么事件準(zhǔn)備就緒
                if(sk.isAcceptable()) {
                    //接受就緒，獲取客戶端連接
                    SocketChannel sc = channel.accept();
                    //設(shè)置非阻塞異步模式
                    sc.configureBlocking(false);
                    //將通道注冊(cè)到服務(wù)器上
                    sc.register(open, SelectionKey.OP_READ);
                } else if(sk.isReadable()) {
                    //獲取當(dāng)前選擇器就緒的通道
                    SocketChannel s =  (SocketChannel) sk.channel();
                    ByteBuffer bb = ByteBuffer.allocate(1024);
                    int len = 0;
                    while((len = s.read(bb)) > 0) {
                        bb.flip();
                        System.out.println(new String(bb.array(),0,len));
                        bb.clear();
                    }
               }
          }
         it.remove();
     }
}