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

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

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

一、概述

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

二、常用概念

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

客戶端實現(xiàn)

步驟

創(chuàng)建SocketChannel 通道
切換異步非阻塞模式configureBlocking(false)
設置緩沖區(qū)大小ByteBuffer.allocate(1024)
值寫入緩沖區(qū) buffer.put(input.getBytes())
緩沖區(qū)中的值寫入通道中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);
      //設置緩沖去大小
      ByteBuffer buffer=ByteBuffer.allocate(1024);
      System.out.println("輸入傳輸值：");
      //獲取鍵盤輸入的值
      Scanner scanner = new Scanner(System.in);
      while (scanner.hasNext()){
          String input=scanner.next();
          //把獲取的值寫入緩沖區(qū)中
          buffer.put(input.getBytes());
          buffer.flip();
          //把緩沖區(qū)中的值寫入通道中
          channel.write(buffer);
          buffer.clear();
       }
       channel.close();
 }

服務端實現(xiàn)

步驟

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

代碼演示

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();
        //將通道注冊到選擇器,并指定監(jiān)聽接受事件
        channel.register(open, SelectionKey.OP_ACCEPT);
        //輪訓式獲取選擇已經(jīng)準備就緒的事件
        while(open.select() > 0) {
            //獲取當前選擇器所有注冊的監(jiān)聽事件
            Iterator<SelectionKey> it = open.selectedKeys().iterator();
            while(it.hasNext()) {
                //獲取準備就緒的事件
                SelectionKey sk = it.next();
                //判斷是什么事件準備就緒
                if(sk.isAcceptable()) {
                    //接受就緒，獲取客戶端連接
                    SocketChannel sc = channel.accept();
                    //設置非阻塞異步模式
                    sc.configureBlocking(false);
                    //將通道注冊到服務器上
                    sc.register(open, SelectionKey.OP_READ);
                } else if(sk.isReadable()) {
                    //獲取當前選擇器就緒的通道
                    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();
     }
}