當前位置：首頁 > 编程资源 > 编程问答 >内容正文

编程问答

非阻塞模式

發(fā)布時間：2025/3/21 编程问答 52 豆豆

生活随笔收集整理的這篇文章主要介紹了非阻塞模式小編覺得挺不錯的,現(xiàn)在分享給大家,幫大家做個參考.

本篇文章觀點和例子來自《Java網(wǎng)絡編程精解》，作者為孫衛(wèi)琴，出版社為電子工業(yè)出版社。

?[轉(zhuǎn)]http://blog.csdn.net/xhh198781/article/details/6635775

????? 對于用ServerSocket 及 Socket 編寫的服務器程序和客戶程序, 他們在運行過程中常常會阻塞. 例如, 當一個線程執(zhí)行 ServerSocket 的accept() 方法時, 假如沒有客戶連接, 該線程就會一直等到有客戶連接才從 accept() 方法返回. 再例如, 當線程執(zhí)行 Socket 的 read() 方法時, 如果輸入流中沒有數(shù)據(jù), 該線程就會一直等到讀入足夠的數(shù)據(jù)才從 read() 方法返回.

????? 假如服務器程序需要同時與多個客戶通信, 就必須分配多個工作線程, 讓他們分別負責與一個客戶通信, 當然每個工作線程都有可能經(jīng)常處于長時間的阻塞狀態(tài).

????? 從 JDK1.4 版本開始, 引入了非阻塞的通信機制. 服務器程序接收客戶連接, 客戶程序建立與服務器的連接, 以及服務器程序和客戶程序收發(fā)數(shù)據(jù)的操作都可以按非阻塞的方式進行. 服務器程序只需要創(chuàng)建一個線程, 就能完成同時與多個客戶通信的任務.

??????非阻塞的通信機制主要由 java.nio 包(新I/O包) 中的類實現(xiàn), 主要的類包括 ServerSocketChannel, SocketChannel, Selector, SelectionKey 和 ByteBuffer 等.

????? 本章介紹如何用 java.nio 包中的類來創(chuàng)建服務器程序和客戶程序, 并且分別采用阻塞模式和非阻塞模式來實現(xiàn)它們. 通過比較不同的實現(xiàn)方式, 可以幫助讀者理解它們的區(qū)別和適用范圍.

一. 線程阻塞的概念

????? 在生活中, 最常見的阻塞現(xiàn)象是公路上汽車的堵塞. 汽車在公路上快速行駛, 如果前方交通受阻, 就只好停下來等待, 等到交通暢順, 才能恢復行駛.

????? 線程在運行中也會因為某些原因而阻塞. 所有處于阻塞狀態(tài)的線程的共同特征是: 放棄CPU, 暫停運行, 只有等到導致阻塞的原因消除, 才能恢復運行; 或者被其他線程中斷, 該線程會退出阻塞狀態(tài), 并且拋出 InterruptedException.

1.1 線程阻塞的原因

????? 導致線程阻塞的原因主要有以下幾方面.

線程執(zhí)行了 Thread.sleep(int n) 方法, 線程放棄 CPU, 睡眠 n 毫秒, 然后恢復運行.
線程要執(zhí)行一段同步代碼, 由于無法獲得相關(guān)的同步鎖, 只好進入阻塞狀態(tài), 等到獲得了同步鎖, 才能恢復運行.
線程執(zhí)行了一個對象的 wait() 方法, 進入阻塞狀態(tài), 只有等到其他線程執(zhí)行了該對象的 notify() 和 notifyAll() 方法, 才可能將其呼醒.
線程執(zhí)行 I/O 操作或進行遠程通信時, 會因為等待相關(guān)的資源而進入阻塞狀態(tài). 例如, 當線程執(zhí)行 System.in.read() 方法時, 如果用戶沒有向控制臺輸入數(shù)據(jù), 則該線程會一直等讀到了用戶的輸入數(shù)據(jù)才從 read() 方法返回.

???? 進行遠程通信時, 在客戶程序中, 線程在以下情況可能進入阻塞狀態(tài).

請求與服務器建立連接時, 即當線程執(zhí)行 Socket 的帶參數(shù)構(gòu)造方法, 或執(zhí)行 Socket 的 connect() 方法時, 會進入阻塞狀態(tài), 直到連接成功, 此線程才從 Socket 的構(gòu)造方法或 connect() 方法返回.
線程從 Socket 的輸入流讀入數(shù)據(jù)時, 如果沒有足夠的數(shù)據(jù), 就會進入阻塞狀態(tài), 直到讀到了足夠的數(shù)據(jù), 或者到達輸入流的末尾, 或者出現(xiàn)了異常, 才從輸入流的 read() 方法返回或異常中斷. 輸入流中有多少數(shù)據(jù)才算足夠呢? 這要看線程執(zhí)行的 read() 方法的類型.

＞ int read(): 只要輸入流中有一個字節(jié), 就算足夠.

＞ int read( byte[] buff): 只要輸入流中的字節(jié)數(shù)目與參數(shù)buff?數(shù)組的長度相同, 就算足夠.

＞ String readLine(): 只要輸入流中有一行字符串, 就算足夠. 值得注意的是, InputStream 類并沒有 readLine() 方法, 在過濾流 BufferedReader 類中才有此方法.?

線程向 Socket 的輸出流寫一批數(shù)據(jù)時, 可能會進入阻塞狀態(tài), 等到輸出了所有的數(shù)據(jù), 或者出現(xiàn)異常, 才從輸出流的 write() 方法返回或異常中斷.
調(diào)用 SOcket 的setSoLinger() 方法設置了關(guān)閉 Socket 的延遲時間, 那么當線程執(zhí)行 Socket 的 close() 方法時, 會進入阻塞狀態(tài),?直到底層 Socket 發(fā)送完所有剩余數(shù)據(jù), 或者超過了 setSoLinger() 方法設置的延遲時間, 才從 close() 方法返回.

??????在服務器程序中, 線程在以下情況下可能會進入阻塞狀態(tài).

線程執(zhí)行 ServerSocket 的 accept() 方法, 等待客戶的連接,?直到接收到了客戶連接, 才從 accept() 方法返回.??????
線程從 Socket 的輸入流讀入數(shù)據(jù)時, 如果輸入流沒有足夠的數(shù)據(jù), 就會進入阻塞狀態(tài).
線程向 Socket 的輸出流寫一批數(shù)據(jù)時, 可能會進入阻塞狀態(tài), 等到輸出了所有的數(shù)據(jù), 或者出現(xiàn)異常, 才從輸出流的 write() 方法返回或異常中斷.

????? 由此可見, 無論在服務器程序還是客戶程序中, 當通過 Socket 的輸入流和輸出流來讀寫數(shù)據(jù)時, 都可能進入阻塞狀態(tài). 這種可能出現(xiàn)阻塞的輸入和輸出操作被稱為阻塞 I/O. 與此對照, 如果執(zhí)行輸入和輸出操作時, 不會發(fā)生阻塞, 則稱為非阻塞 I/O.

1.2 服務器程序用多線程處理阻塞通信的局限

????? 本書第三章的第六節(jié)(創(chuàng)建多線程的服務器) 已經(jīng)介紹了服務器程序用多線程來同時處理多個客戶連接的方式. 服務器程序的處理流程如圖 4-1 所示. 主線程負責接收客戶的連接. 在線程池中有若干工作線程, 他們負責處理具體的客戶連接. 每當主線程接收到一個客戶連接, 就會把與這個客戶交互的任務交給一個空閑的工作線程去完成, 主線程繼續(xù)負責接收下一個客戶連接.

????????????????????????????? 圖4-1 服務器程序用多線程處理阻塞通信

????? 在圖4-1 總, 用粗體框標識的步驟為可能引起阻塞的步驟. 從圖中可以看出, 當主線程接收客戶連接, 以及工作線程執(zhí)行 I/O 操作時, 都有可能進入阻塞狀態(tài).

????? 服務器程序用多線程來處理阻塞 I/O, 盡管能滿足同時響應多個客戶請求的需求, 但是有以下局限:

????? ⑴ Java 虛擬機會為每個線程分配獨立的堆棧空間, 工作線程數(shù)目越多, 系統(tǒng)開銷就越大, 而且增加了 Java虛擬機調(diào)度線程的負擔, 增加了線程之間同步的復雜性, 提高了線程死鎖的可能性;

??????⑵ 工作線程的許多時間都浪費在阻塞 I/O 操作上, Java 虛擬機需要頻繁地轉(zhuǎn)讓 CPU 的使用權(quán), 使進入阻塞狀態(tài)的線程放棄CPU, 再把CPU 分配給處于可運行狀態(tài)的線程.

????? 由此可見, 工作線程并不是越多越好. 如圖 4-2 所示, 保持適量的工作線程, 會提高服務器的并發(fā)性能, 但是當工作線程的數(shù)目達到某個極限, 超出了系統(tǒng)的負荷時, 反而會減低并發(fā)性能, 使得多數(shù)客戶無法快速得到服務器的響應.

????????????????????? 圖4-2 線程數(shù)目與并發(fā)性能的更新??????????????????

1.3 非阻塞通信的基本思想

????? 假如要同時做兩件事: 燒開水和燒粥. 燒開水的步驟如下:

????? 鍋里放水, 打開煤氣爐;

????? 等待水燒開;??????????????????????????????????????????????????????????? //阻塞

????? 關(guān)閉煤氣爐, 把開水灌到水壺里;

??????燒粥的步驟如下:

????? 鍋里放水和米, 打開煤氣爐;

????? 等待粥燒開;???????????????????????????????????????????????????????????? //阻塞

????? 調(diào)整煤氣爐, 改為小火;???

????? 等待粥燒熟;???????????????????????????????????????????????????????????? //阻塞

??????關(guān)閉煤氣爐;

??????為了同時完成兩件事, 一個方案是同時請兩個人分別做其中的一件事, 這相當于采用多線程來同時完成多個任務. 還有一種方案是讓一個人同時完成兩件事, 這個人應該善于利用一件事的空閑時間去做另一件事, 一刻也不應該閑著:

??????鍋子里放水, 打開煤氣爐;????????????????????? //開始燒水

????? 鍋子力放水和米, 打開煤氣爐;??????????????? //開始燒粥

????? while(一直等待, 直到有水燒開, 粥燒開或粥燒熟事件發(fā)生){????????? //阻塞

??????????? if(水燒開)

???????????????????關(guān)閉煤氣爐, 把開水灌到水壺里;

??????????? if(粥燒開)

?????????????????? 調(diào)整煤氣爐, 改為小火;

??????????? if(粥燒熟)

?????????????????? 關(guān)閉煤氣爐;

????????????if(水已經(jīng)燒開并且粥已經(jīng)燒熟)

?????????????????? 退出循環(huán);?

????? }?????????//這里的煤氣爐我可以理解為每件事就有一個煤氣爐配給吧, 這也是一部分的開銷呢

???????????????? //并且if里面的動作必須要能快速完成的才行, 不然后面的就要排隊了

???????????????? //如是太累的工作還是不要用這個好???????????????????????????????????

????? 這個人不斷監(jiān)控燒水及燒粥的狀態(tài), 如果發(fā)生了 "水燒開", "粥燒開" 或 "粥燒熟" 事件, 就去處理這些事件, 處理完一件事后進行監(jiān)控燒水及燒粥的狀態(tài), 直到所有的任務都完成.

?????? 以上工作方式也可以運用到服務器程序中, 服務器程序只需要一個線程就能同時負責接收客戶的連接, 接收各個客戶發(fā)送的數(shù)據(jù), 以及向各個客戶發(fā)送響應數(shù)據(jù). 服務器程序的處理流程如下:

?????? while(一直等待, 直到有接收連接就緒事件, 讀就緒事件或?qū)懢途w事件發(fā)生){???????????? //阻塞

????????????? if(有客戶連接)

?????????????????? 接收客戶的連接;??????????????????????????????????????????????????? //非阻塞

????????????? if(某個 Socket 的輸入流中有可讀數(shù)據(jù))

?????????????????? 從輸入流中讀數(shù)據(jù);???????????????????????????????????????????????? //非阻塞

??????????????if(某個 Socket 的輸出流可以寫數(shù)據(jù))

???????????????????向輸出流寫數(shù)據(jù);??????????????????????????????????????????????????? //非阻塞

?????? }

??????以上處理流程采用了輪詢的工作方式, 當某一種操作就緒時, 就執(zhí)行該操作, 否則就查看是否還有其他就緒的操作可以執(zhí)行. 線程不會因為某一個操作還沒有就緒, 就進入阻塞狀態(tài), 一直傻傻地在那里等待這個操作就緒.

??????為了使輪詢的工作方式順利進行, 接收客戶的連接, 從輸入流讀數(shù)據(jù), 以及向輸出流寫數(shù)據(jù)的操作都應該以非阻塞的方式運行. 所謂非阻塞, 就是指當線程執(zhí)行這些方法時, 如果操作還沒有就緒, 就立即返回, 而不會一直等到操作就緒. 例如, 當線程接收客戶連接時, 如果沒有客戶連接, 就立即返回; 再例如, 當線程從輸入流中讀數(shù)據(jù)時, 如果輸入流中還沒有數(shù)據(jù), 就立即返回, 或者如果輸入流還沒有足夠的數(shù)據(jù), 那么就讀取現(xiàn)有的數(shù)據(jù), 然后返回. 值得注意的是, 以上 while 學校條件中的操作還是按照阻塞方式進行的, 如果未發(fā)生任何事件, 就會進入阻塞狀態(tài), 直到接收連接就緒事件, 讀就緒事件或?qū)懢途w事件中至少有一個事件發(fā)生時, 才會執(zhí)行 while 循環(huán)體中的操作. 在while 循環(huán)體中, 一般會包含在特定條件下退出循環(huán)的操作.

二. java.nio 包中的主要類

??????? java.nio 包提供了支持非阻塞通信的類.

ServerSocketChannel: ServerSocket 的替代類, 支持阻塞通信與非阻塞通信.
SocketChannel: Socket 的替代類, 支持阻塞通信與非阻塞通信.
Selector: 為ServerSocketChannel 監(jiān)控接收連接就緒事件, 為 SocketChannel 監(jiān)控連接就緒, 讀就緒和寫就緒事件.
SelectionKey: 代表 ServerSocketChannel 及 SocketChannel 向 Selector 注冊事件的句柄. 當一個 SelectionKey 對象位于Selector 對象的 selected-keys 集合中時, 就表示與這個 SelectionKey 對象相關(guān)的事件發(fā)生了.??????

??????ServerSocketChannel 及 SocketChannel 都是 SelectableChannel 的子類, 如圖 4-3 所示. SelectableChannel 類及其子類都能委托 Selector 來監(jiān)控他們可能發(fā)生的一些事件, 這種委托過程也稱為注冊事件過程.

??????????????????????????????

?????????????????????????????????????????? 圖4-3 SelectableChannel 類及其子類的類框圖

??????ServerSocketChannel 向 Selector 注冊接收連接就緒事件的代碼如下:

????????????SelectionKey key = serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);?????????????????????????

????? SelectionKey 類的一些靜態(tài)常量表示事件類型, ServerSocketChannel 只可能發(fā)生一種事件.

SelectionKey.OP_ACCEPT: 接收連接就緒事件, 表示至少有了一個客戶連接, 服務器可以接收這個連接.

?????SocketChannel 可能發(fā)生以下 3 種事件.

SelectionKey.OP_CONNECT: 連接就緒事件, 表示客戶與服務器的連接已經(jīng)建立成功.
SelectionKey.OP_READ: 讀就緒事件,? 表示輸入流中已經(jīng)有了可讀數(shù)據(jù), 可以執(zhí)行讀操作了
SelectionKey.OP_WRITE: 寫就緒事件, 表示已經(jīng)可以向輸入流寫數(shù)據(jù)了.

???? SocketChannel 提供了接收和發(fā)送數(shù)據(jù)的方法.

read(ByteBuffer buffer): 接收數(shù)據(jù), 把它們存放到參數(shù)指定的 ByteBuffer 中.
write(ByteBuffer buffer): 把參數(shù)指定的 ByteBuffer 中的數(shù)據(jù)發(fā)送出去.

???? ByteBuffer 表示字節(jié)緩沖區(qū), SocketChannel 的 read() 和 write() 方法都會操縱 ByteBuffer. ByteBuffer 類繼承于 Buffer 類. ByteBuffer 中存放的是字節(jié),? 為了把它們轉(zhuǎn)換為字符串, 還需要用到 Charset 類, Charset 類代表字符編碼, 它提供了把字節(jié)流轉(zhuǎn)換為字符串(解碼過程) 和把字符串轉(zhuǎn)換為字節(jié)流(編碼過程) 的實用方法.

?????? 下面幾小節(jié)分別介紹 Buffer, Charset, Channel, SelectableChannel, ServerSocketChannel, SocketChannel, Selector 和 SelectionKey 的用法.

2.1 緩沖區(qū) Buffer

??????數(shù)據(jù)輸入和輸出往往是比較耗時的操作. 緩沖區(qū)從兩個方面提高 I/O 操作的效率:

減少實際的物理讀寫次數(shù);
緩存區(qū)在創(chuàng)建時被分配內(nèi)存, 這塊內(nèi)存區(qū)域一直被重用, 這可以減少動態(tài)分配和回收內(nèi)存區(qū)域的次數(shù).

????? 舊I/O 類庫(對應 java.nio包) 中的 BufferedInputStream, BufferedOutputStream, BufferedReader 和 BufferedWriter 在其實現(xiàn)中都運用了緩沖區(qū). java.nio 包公開了 Buffer API, 使得Java 程序可以直接控制和運用緩沖區(qū). 如圖 4-4 所示, 顯示了 Buffer 類的層次結(jié)構(gòu).

?????????????????

?????????????????????????????????????????圖 4-4 Buffer 類的層次結(jié)構(gòu)

?????所有的緩沖區(qū)都有以下屬性:

容量(capacity): 表示該緩沖區(qū)可以保存多少數(shù)據(jù)．
極限(limit): 表示緩沖區(qū)的當前終點, 不能對緩沖區(qū)中超過極限的區(qū)域進行讀寫操作. 極限是可以修改的, 這有利于緩沖區(qū)的重用. 例如, 假定容量100 的緩沖區(qū)已經(jīng)填滿了數(shù)據(jù), 接著程序在重用緩沖區(qū)時, 僅僅將 10 個新的數(shù)據(jù)寫入緩沖區(qū)中從位置0 到10 的區(qū)域, 這時可以將極限設為 10, 這樣就不能讀取先前的數(shù)據(jù)了. 極限是一個非負整數(shù), 不應該大于容量.
位置(position): 表示緩沖區(qū)中下一個讀寫單元的位置, 每次讀寫緩沖區(qū)的數(shù)據(jù)時, 都會改變該值, 為下一次讀寫數(shù)據(jù)作準備. 位置是一個非負整數(shù), 不應該大于極限.

???? 如圖 4-5 所示, 以上 3 個屬性的關(guān)系為: 容量 ≥ 極限 ≥ 位置 ≥ 0

???????????????????

????????????????????圖 4-5 緩沖區(qū)的 3 個屬性

???? 緩沖區(qū)提供了用于改變以上 3 個屬性的方法.

clear(): 把極限設為容量, 再把位置設為 0;
flip(): 把極限設為位置, 再把位置設為 0;
rewind(): 不改變極限, 把位置設為 0.

????? Buffer 類的remaining() 方法返回緩沖區(qū)的剩余容量, 取值等于極限-位置. Buffer 類的 compact() 方法刪除緩沖區(qū)內(nèi)從 0 到當前位置position 的內(nèi)容, 然后把從當前位置position 到極限limit 的內(nèi)容復制到 0 到 limit-position 的區(qū)域內(nèi), 當前位置position 和極限limit 的取值也作相應的變化, 如圖 4-6 所示.

?????????????????????

????????????????????????????????? 圖4-6 Buffer 類的 compact() 的作用

???????? java.nio.Buffer 類是一個抽象類, 不能被實例化. 共有 8 個具體的緩沖區(qū)類, 其中最基本的緩沖區(qū)是 ByteBuffer, 它存放的數(shù)據(jù)單元是字節(jié). ByteBuffer 類并沒有提供公開的構(gòu)造方法, 但是提供了兩個獲得 ByteBuffer 實例的靜態(tài)工廠方法.

allocate(int capacity):?返回一個 ByteBuffer 對象, 參數(shù)capacity 指定緩沖區(qū)的容量.
directAllocate(int capacity):返回一個 ByteBuffer 對象, 參數(shù)capacity 指定緩沖區(qū)的容量. 該方法返回的緩沖區(qū)稱為直接緩沖區(qū), 它與當前操作系統(tǒng)能夠更好地耦合, 因此能進一步提高 I/O 操作的速度. 但是直接分配緩沖區(qū)的系統(tǒng)開銷很大, 因此只有在緩沖區(qū)較大并且長期存在, 或者需要經(jīng)常重用時, 才使用這種緩沖區(qū).

????? 除 boolean 類型以外, 每種基本類型都有對應的緩沖區(qū)類, 包括 CharBuffer, DoubleBuffer, FloatBuffer, IntBuffer, LongBuffer 和 ShortBuffer. 這幾個緩沖區(qū)類都有一個能夠返回自身實例的靜態(tài)工廠方法allocate(int capacity). 在 CharBuffer 中存放的數(shù)據(jù)單元為字符, 在 DoubleBuffer 中存放的數(shù)據(jù)單元為 double 數(shù)據(jù), 依此類推. 還有一個緩沖區(qū)是 MappedByteBuffer, 它是 ByteBuffer 的子類. MappedByteBuffer 能夠把緩沖區(qū)和文件的某個區(qū)域直接映射.

??????所有具有緩沖區(qū)類都提供了讀寫緩沖區(qū)的方法:

get(): 相對讀. 從緩沖區(qū)的當前位置讀取一個單元的數(shù)據(jù), 讀完后把位置加 1;
get(int index): 絕對讀. 從參數(shù) index 指定的位置讀取一個單元的數(shù)據(jù);
put(): 相對寫. 向緩沖區(qū)的當前位置寫入一個單元的數(shù)據(jù), 寫完后把位置加 1;
put(int index): 絕對寫. 向參數(shù) index 指定的位置寫入一個單元的數(shù)據(jù).

2.2? 字符編碼 Charset

??????java.nio.Channel 類的每個實例代表特定的字符編碼類型. 如圖 4-7 所示, 把字節(jié)序列轉(zhuǎn)換為字符串的過程稱為解碼; 把字符串轉(zhuǎn)換為字節(jié)序列的過程稱為編碼.

??????

???????????????????????????? 圖 4-7 編碼與解碼

??????Charset 類提供了編碼與解碼的方法:

ByteBuffer encode(String str): 對參數(shù) Str 指定的字符串進行編碼, 把得到的字節(jié)序列存放在一個 ByteBuffer 對象中, 并將其返回;
ByteBuffer encode(CharBuffer cb): 對參數(shù) cb 指定的字符緩沖區(qū)中的字符進行編碼,把得到的字節(jié)序列存放在一個 ByteBuffer 對象中, 并將其返回;
CharBuffer decode(ByteBuffer bb):　把參數(shù) bb 指定的 ByteBuffer 中的字節(jié)序列進行解碼, 把得到的字符序列存放在一個 CharBuffer 對象中, 并將其返回.

????? Charset 類的靜態(tài) forName(String encode) 方法返回一個 Charset 對象, 它代表參數(shù) encode 指定的編碼類型. 例如, 以下代碼創(chuàng)建了一個代表"GBK" 編碼的 Charset對象:

???????? Charset charset = Charset.forName("GBK");???????????????????????????????????????

?????? Charset?類還有一個靜態(tài)方法 defaultCharset(), 它返回代表本地平臺的默認字符編碼的 Charset 對象.?

2.3 通道Channel

????? 通道 Channel 用來連接緩沖區(qū)與數(shù)據(jù)源或數(shù)據(jù)匯(數(shù)據(jù)目的地). 如圖4-8 所示, 數(shù)據(jù)源的數(shù)據(jù)經(jīng)過管道到達緩沖區(qū), 緩沖區(qū)的數(shù)據(jù)經(jīng)過通道到達數(shù)據(jù)匯.

??????

??????????????????????????????????? 圖4-8 通道的作用

???????如圖 4-9 所示, 顯示了 Channel 的主要層次結(jié)構(gòu).

???????

??????????????????????????????????????????????????????? 圖4-9 Channel 的主要層次結(jié)構(gòu)

?????? java.nio.channels.Channel 接口只聲明了兩個方法.

close(): 關(guān)閉通道;
isOpen(): 判斷通道是否打開.

????? 通道在創(chuàng)建時被打開, 一旦關(guān)閉通道, 就不能重新打開了.

????? Channel 接口的兩個最重要的子接口是 ReadableByteChannel 和 WritableByteChannel. ReadableByteChannel 接口聲明了 read(ByteBuffer dst) 方法, 該方法把數(shù)據(jù)源的數(shù)據(jù)讀入?yún)?shù)指定的 ByteBuffer 緩沖區(qū)中; WritableByteChannel 接口聲明了 write(ByteBuffer src)方法, 該方法把參數(shù)指定的 ByteBuffer 緩沖區(qū)中的數(shù)據(jù)寫到數(shù)據(jù)匯中. 如圖4-10 所示, 顯示了 Channel 與 Buffer 的關(guān)系. ByteChannel 接口是一個便利接口, 它擴展了 ReadByteChannel 和 WritableByteChannel 接口, 因而同時支持讀寫操作.

???????

???????????????????????????? 圖4-10 Channel 與 Buffer ? 的關(guān)系

??????ScatteringByteChannel 接口擴展了 ReadByteChannel 接口, 允許分散地讀取數(shù)據(jù). 分散讀取數(shù)據(jù)是指單個讀取操作能填充多個緩沖區(qū).? ScatteringByteChannel 接口聲明了 read(ByteBuffer[] dsts)方法, 該方法把從數(shù)據(jù)源讀取的數(shù)據(jù)依次填充到參數(shù)指定的 ByteBuffer 數(shù)組的各個 ByteBuffer 中. GatheringByteChannel 接口擴展了 WritableByteChannel 接口, 允許集中地寫入數(shù)據(jù). 集中寫入數(shù)據(jù)是指單個寫操作能把多個緩沖區(qū)的數(shù)據(jù)寫入數(shù)據(jù)匯. GatheringByteChannel 接口聲明了 write(ByteBuffer[] srcs)方法, 該方法依次把參數(shù)指定的 ByteBuffer 數(shù)組的每個 ByteBuffer 中的數(shù)據(jù)寫入數(shù)據(jù)匯. 分散讀取和集中寫數(shù)據(jù)能夠進一步提高輸入和輸出操作的速度.

???? FileChannel 類是 Channel 接口的實現(xiàn)類, 代表一個與文件相連的通道. 該類實現(xiàn)了 ByteChannel, ScatteringByteChannel, GatheringByteChannel 接口, 支持讀操作, 寫操作, 分散讀操作和集中寫操作. FileChannel 類沒有提供公開的構(gòu)造方法, 一次客戶程序不能用 new 語句來構(gòu)造它的實現(xiàn). 不過, 在 FileInputStream, FileOutputStream 和 RandomAccessFile 類中提供了 getChannel() 方法, 該方法返回對應的 FileChannel 對象.

????? SelectableChannel 也是一種通道, 它不僅支持阻塞的 I/O 操作, 還支持非阻塞的 I/O 操作. SelectableChannel 有兩個子類: ServerSocketChannel 和 SocketChannel. SocketChannel 還實現(xiàn)了 ByteChannel 接口, 具有 read(ByteBuffer dst) 和 write(ByteBuffer src) 方法.

?????? 注意上面的圖4-9 Channel 的主要層次結(jié)構(gòu), 這個跟原書有點區(qū)別, 里面的類都是 jdk1.5的, 其中 SocketChannel 是實現(xiàn)了 ByteChannel, ScatteringByteChannel, GatheringByteChannel 接口, SocketChannel 還有一個子類SocketChannelImpl, SocketChannelImpl 的源代碼看不到呢.

2.4 SelectableChannel 類

??????SelectableChannel 類是一種支持阻塞 I/O 和非阻塞 I/O 的通道. 在非阻塞模式下, 讀寫數(shù)據(jù)不會阻塞, 并且SelectableChannel 可以向 Selector 注冊讀就緒和寫就緒等事件. Selector 負責監(jiān)控這些事件, 等到事件發(fā)生時, 比如發(fā)生了讀就緒事件, SelectableChannel 就可以執(zhí)行讀操作了.

????? SelectableChannel 的主要方法如下:

public SelecotableChannel configureBlocking(boolean block) throws IOException

???? 數(shù)block 為true 時, 表示把 SelectableChannel 設為阻塞模式; 如果參數(shù)block 為false, 表示把 SelectableChannel 設為非阻塞模式. 默認情況下, SelectableChannel 采用阻塞模式. 該方法返回 SelectableChannel 對象本身的引用, 相當于" return this".

public SelectionKey register(Selector sel, int ops) throws ClosedChannelException
public SelectionKey register(Selector sel, int ops, Object attachment) throws ClosedChannelException

????? 后兩個方法都向 Selector 注冊時間, 如以下 socketChannel( SelectableChannel 的一個子類) 向 Selector 注冊讀就緒和寫就緒事件:

???????? SelectionKey key = socketChannel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE);

?????? register() 方法返回一個 SelectionKey 對象, SelectionKey 用來跟蹤被注冊的事件.?第二個register() 方法還有一個Object 類型的參數(shù) attachment, 它用于為 SelectionKey 關(guān)聯(lián)一個附件, 當被注冊事件發(fā)生后, 需要處理該事件時, 可以從 SelectionKey 中獲得這個附件, 該附件可用來包含與處理這個事件相關(guān)的信息. 以下這兩段代碼是等價的:

等價于:

??????? MyHandler handler = new MyHandler();???????????? //負責處理事件的對象
??????? SelectionKey key = socketChannel.register(selector, SelectioinKey.OP_READ | SelectionKey.OP_WRITE);
??????? key.attach(handler?);?????????????????????????????????????????? //為SelectionKey 關(guān)聯(lián)一個附件

?2.5 ServerSocketChannel 類

?????? SeverSocketChannel 從 SeletableChannel 中繼承了 configureBlocking() 和 register()方法. ServerSocketChannel 是 ServerSocket 的替換類, 也具有負責接收客戶連接的 accept() 方法. ServerSocket 并沒有 public 類型的構(gòu)造方法, 必須通過它的靜態(tài)方法open() 來創(chuàng)建 ServerSocketChannel 對象. 每個ServerSocketChannel 對象都與一個ServerSocket 對象關(guān)聯(lián). ServerSocketChannel 的 socket() 方法返回與它關(guān)聯(lián)的 ServerSocket 對象. 可通過以下方法把服務器進程綁定到一個本地端口:

???????????serverSocketChannel.socket().bind(port);??????????????????????????????????????????????????????????

?????? ServerSocketChannel 的主要方法如下:

public static ServerSocketChannel open() throws IOException

?????這是 ServerSocketChannel 類的靜態(tài)工廠方法, 它返回一個 ServerSocketChannel 對象, 這個對象沒有與任何本地端口綁定, 并且處于阻塞模式.

public SocketChannel accept() throws IOException

????? 類似于 ServerSocket 的accept() 方法, 用于接收客戶的連接. 如果 ServerSocketChannel?處于非阻塞狀態(tài), 當沒有客戶連接時, 該方法立即返回 null; 如果ServerSocketChannel 處于阻塞狀態(tài), 當沒有客戶連接時, 它會一直阻塞下去,?直到有客戶連接就緒, 或者出現(xiàn)了IOException.

????? 值得注意的是, 該方法返回的 SocketChannel 對象處于阻塞模式, 如果希望把它改為非阻塞模式, 必須執(zhí)行以下代碼:

??????? socketChannel.configureBlocking(false);???????????????????????????????????????

public final int validOps()

?????返回 ServerSocketChannel 所能產(chǎn)生的事件, 這個方法總是返回 SelectionKey.OP_ACCEPT.

public ServerSocket socket()

???? 返回與 ServerSocketChannel 關(guān)聯(lián)的 ServerSocket 對象. 每個 ServerSocketChannel 對象都與一個 ServerSocket 對象關(guān)聯(lián).

2.6 SocketChannel 類

??????SocketChannel 可看作是 Socket 的替代類, 但它比 Socket 具有更多的功能. SocketChannel 不僅從 SelectableChannel 父類中繼承了 configureBlocking() 和 register() 方法, 并且實現(xiàn)了 ByteChannel 接口, 因此具有用于讀寫數(shù)據(jù)的 read(ByteBuffer dst) 和 write(ByteBuffer src) 方法. SocketChannel 沒有public 類型的構(gòu)造方法, 必須通過它的靜態(tài)方法open() 來創(chuàng)建 SocketChannel 對象.

????? SocketChannel 的主要方法如下:

public static SocketChannel open() throws IOException
public static SocketChannel open(SocketAddress remote) throws IOException

????? SocketChannel 的靜態(tài)工廠方法open() 負責創(chuàng)建 SocketChannel 對象, 第二個帶參數(shù)的構(gòu)造方法還會建立與遠程服務器的連接. 在阻塞模式下及非阻塞模式下, 第二個open() 方法有不同的行為, 這與 SocketChannel 類的 connect() 方法類似, 可參見本屆 connect() 方法的介紹.

????? 以下兩段代碼是等價的:

??????? SocketChannel socketChannel = SocketChannel.open();
??????? socketChannel.connect(remote);?????? //remote 為 SocketAddress 類型

????? 等價于:

??????? SocketChannel socketChannel = SocketChannel.open(remote); //remote 為 SocketAddress 類型

????? 值得注意的是, open() 方法返回的SocketChannel 對象處于阻塞模式, 如果希望把它改為非阻塞模式, 必須執(zhí)行以下代碼:

???????? socketChannel.configureBlock(false);

public final int validOps()

????? 返回SocketChannel 所能產(chǎn)生的事件, 這個方法總是返回以下值:

??????? SelectionKey.OP_CONNECT | SelectionKey.OP_READ | SelectionKey.OP_WRITEN????????????????????????

public Socket socket()

????? 返回與這個SocketChannel 關(guān)聯(lián)的 Socket 對象. 每個 SocketChannel 對象都與一個 Socket 對象關(guān)聯(lián).

public boolean isConnected()

????? 判斷底層 Socket 是否已經(jīng)建立了遠程連接.

public boolean isConnectionPending()

????? 判斷是否正在進行遠程連接. 當遠程連接操作已經(jīng)開始, 但是還沒有完成時, 則返回true, 否則返回false. 也就是說, 當?shù)讓覵ocket 還沒有開始連接, 或者已經(jīng)連接成功時, 該方法都會返回false.

public boolean connect(SocketAddress remote)throws IOException

????? 使底層Socket 建立遠程連接. 當SocketChannel 處于非阻塞模式時, 如果立即連接成功, 該方法返回true, 如果不能立即連接成功, 該方法返回false, 程序過會兒必須通過調(diào)用finishConnect() 方法來完成連接. 當SocketChannel 處于阻塞模式, 如果立即連接成功, 該方法返回true, 如果不能立即連接成功, 將進入阻塞狀態(tài), 直到連接成功, 或者出現(xiàn) I/O 異常.

public boolean finishConnect() throws IOExcetion

????? 試圖完成連接遠程服務器的操作. 在非阻塞模式下, 建立連接從調(diào)用SocketChannel 的connect() 方法開始, 到調(diào)用 finishConnect() 方法結(jié)束. 如果finishConnect() 方法順利完成連接, 或者在調(diào)用次方法之前連接已經(jīng)建立, 則finishConnect() 方法立即返回true. 如果連接操作還沒有完成, 則立即返回false; 如果連接操作中遇到異常而失敗, 則拋出響應的I/O 異常.

????? 在阻塞模式下, 如果連接操作還沒有完成, 則會進入阻塞狀態(tài), 直到連接完成或者出現(xiàn)I/O 異常.

public int read(ByteBuffer dst) throws IOException

????? 從 Channel 中讀入若干字節(jié), 把他們存放到參數(shù)指定的 ByteBuffer 中. 假定執(zhí)行read() 方法前, ByteBuffer 的位置為p, 剩余容量為r, r 等于dst.remaining() 方法的返回值. 假定read() 方法實際讀入了 n 個字節(jié), 那么 0 ≤ n ≤ r. read() 方法返回后, 參數(shù) dst 引用的ByteBuffer 的位置變?yōu)?p+n, 極限保持不變, 如圖4-11 所示:

??????

?????????????????????????????? 圖4-11 read() 方法讀入 n 個字節(jié)

??????在阻塞模式下, read() 方法會爭取讀到 r 個字節(jié), 如果輸入流中不足 r 個字節(jié), 就進入阻塞狀態(tài), 直到讀入了 r 個字節(jié), 或者讀到了輸入流末尾, 或者出現(xiàn)了 I/O 異常.

????? 在非阻塞模式下, read() 方法奉行能讀到多少數(shù)據(jù)就讀多少數(shù)據(jù)的原則. read() 方法讀取當前通道中的可讀數(shù)據(jù), 有可能不足 r 個資金額, 或者為 0 個字節(jié), read() 方法總是立即返回, 而不會等到讀取了 r 個字節(jié)在返回.

??????read() 方法返回的實際上讀入的字節(jié)數(shù), 有可能為 0. 如果返回 -1, 就表示讀到了輸入流的末尾.

public int write(ByteBuffer src) throws IOException

????? 把參數(shù) src 指定的 ByteBuffer 中的字節(jié)寫到 Channel 中. 假定執(zhí)行 write() 方法前, ByteBuffer? 的位置為 p, 剩余容量為 r, r 等于 src.remaining() 方法的返回值. 假定 write() 方法實際上向通道中寫了 n 個字節(jié), 那么 0 ≤ n ≤ r. write() 方法返回后, 參數(shù) src 引用的 ByteBuffer 的位置變?yōu)?p+n, 極限保持不變, 如圖4-12 所示:

??????

???????????????????????????????????????? 圖4-12 write() 方法輸出 n 個字節(jié)

????? 在阻塞模式下, write() 方法會爭取輸出 r 個字節(jié), 如果底層網(wǎng)絡的輸出緩沖區(qū)不能容納 r 個字節(jié), 就進入阻塞狀態(tài), 直到輸出了 r 個字節(jié), 或者出現(xiàn)了 I/O 異常.

????? 在非阻塞模式下, write() 方法奉行能輸出多少數(shù)據(jù)就輸出多少數(shù)據(jù)的原則, 有可能不足 r 個字節(jié), 或者為 0 個字節(jié), write() 方法總是立即返回, 而不會等到輸出 r 個字節(jié)后再返回.

??????write() 方法返回實際上輸出的字節(jié)數(shù), 有可能為 0.

2.7 Selector 類

????? 只要 ServerSocketChannel 及 SocketChannel 向 Selector 注冊了特定的事件, Selector 就會監(jiān)控這些事件是否發(fā)生. SelectableChannel 的 register() 方法負責注冊事件, 該方法返回一個SelectionKey 對象, 該對象是用于跟蹤這些被注冊事件的句柄. 一個Selector 對象中會包含 3 種類型的 SelectionKey 集合.

all-keys 集合: 當前所有向Selector 注冊的 SelectionKey 的集合, Selector 的keys() 方法返回該集合.
selected-keys?集合: 相關(guān)時間已經(jīng)被Selector 捕獲的SelectionKey 的集合. Selector 的selectedKeys()?方法返回該集合.
cancelled-keys 集合: 已經(jīng)被取消的 SelectionKey 的集合. Selector 沒有提供訪問這種集合的方法.

???? ?以上第二種和第三種集合都是第一種集合的子集. 對于一個新建的Selector 對象, 它的上述集合都為空.

??????當執(zhí)行SelectableChannel 的 register() 方法時, 該方法新建一個 SelectionKey, 并把它加入到 Selector 的all-keys 集合中.

????? 如果關(guān)閉了與SelectionKey 對象關(guān)聯(lián)的 Channel 對象, 或者調(diào)用了 SelectionKey 對象的cancel() 方法, 這個 SelectionKey 對象就會被加入到 cancelled-keys 集合中, 表示這個 SelectionKey 對象已經(jīng)被取消, 在程序下一次執(zhí)行 Selector 的 select() 方法時, 被取消的 SelectionKey 對象將從所有的集合(包括 all-keys 集合, selected-keys集合和cancelled-keys 集合)中刪除.

????? 在執(zhí)行 Selector 的 select() 方法時, 如果與 SelectionKey 相關(guān)的事件發(fā)生了, 這個SelectionKey 就被加入到 selected-keys 集合中. 程序直接調(diào)用 selected-keys 集合的 remove() 方法, 或者調(diào)用它的 Iterator 的 remove() 方法, 都可以從 selected-keys 集合中刪除一個 SelectionKey 對象.

????? 程序不允許直接通過集合接口的 remove() 方法刪除 all-keys 集合中的 SelectionKey 對象. 如果程序試圖這樣做, 那么會導致 UnsupportedOperationException.?(all-keys 應該是一個內(nèi)部類, 并且不實現(xiàn)remove()的方法, 繼承的對象也是沒實現(xiàn)這些方法的; 還有可能是new HashSet(){重寫remove()方法,直接拋出異常},如:?private static HashSet keys = new HashSet(){
??public boolean remove(Object o){
???throw new UnsupportedOperationException();
??}
?};)

????? Selector 類的主要方法如下:

?public static Selector open() throws IOException

????? 這是 Selector 的靜態(tài)工廠方法, 創(chuàng)建一個 Selector 對象.

public boolean isOpen()

????? 判斷 Selector 是否處于打開狀態(tài). Selector 對象創(chuàng)建后就處于打開狀態(tài), 當調(diào)用那個了 Selector 對象的 close() 方法, 它就進入關(guān)閉狀態(tài).

public Set<SelectionKey> keys()

??????返回 Selector 的 all-keys 集合, 它包含了所有與 Selector 關(guān)聯(lián)的 SelectionKey 對象.

public int?selectNow() throws IOException

????? 返回相關(guān)事件已經(jīng)發(fā)生的 SelectionKey 對象的數(shù)目. 該方法采用非阻塞的工作方式, 返回當前相關(guān)時間已經(jīng)發(fā)生的 SelectionKey 對象的數(shù)目, 如果沒有, 就立即返回 0 .

public int select() throws IOException
public int select(long timeout) throws IOException

?????該方法采用阻塞的工作方式, 返回相關(guān)事件已經(jīng)發(fā)生的 SelectionKey 對象的數(shù)目, 如果一個也沒有, 就進入阻塞狀態(tài), 直到出現(xiàn)以下情況之一, 才從 select() 方法中返回.

＞至少有一個 SelectionKey 的相關(guān)事件已經(jīng)發(fā)生;

＞其他線程調(diào)用了 Selector 的 wakeup() 方法, 導致執(zhí)行 select() 方法的線程立即從 select() 方法中返回.

＞當前執(zhí)行 select() 方法的線程被其他線程中斷.

＞超出了等待時間. 該時間由 select(long timeout) 方法的參數(shù) timeout 設定, 單位為毫秒. 如果等待超時, 就會正常返回, 但不會拋出超時異常. 如果程序調(diào)用的是不帶參數(shù)的 select() 方法, 那么永遠不會超時, 這意味著執(zhí)行 select) 方法的線程進入阻塞狀態(tài)后, 永遠不會因為超時而中斷.

?public Selector wakeup()

?????呼醒執(zhí)行 Selector 的 select() 方法(也同樣設用于 select(long timeout) 方法) 的線程. 當線程A 執(zhí)行 Selector 對象的 wakeup() 方法時, 如果線程B 正在執(zhí)行同一個 Selector 對象的 select() 方法, 或者線程B 過一會兒會執(zhí)行這個 Selector 對象的 select() 方法, 那么線程B 在執(zhí)行 select() 方法時, 會立即從 select() 方法中返回, 而不會阻塞. 假如, 線程B 已經(jīng)在 select() 方法中阻塞了, 也會立即被呼醒, 從select() 方法中返回.

????? wakeup() 方法只能呼醒執(zhí)行select() 方法的線程B 一次. 如果線程B 在執(zhí)行 select() 方法時被呼醒后, 以后在執(zhí)行 select() 方法, 則仍舊按照阻塞方式工作, 除非線程A 再次調(diào)用 Selector 對象的 wakeup() 方法.

public void close() throws IOException

????? 關(guān)閉 Selector. 如果有其他線程正執(zhí)行這個Selector 的select() 方法并且處于阻塞狀態(tài), 那么這個線程會立即返回. close() 方法使得 Selector 占用的所有資源都被釋放, 所有與 Selector 關(guān)聯(lián)的 SelectionKey 都被取消.

2.8 SelectionKey 類

???? ServerSocketChannel 或 SocketChannel 通過 register() 方法向 Selector 注冊事件時, register() 方法會創(chuàng)建一個 SelectionKey 對象, 這個 SelectionKey 對象是用來跟蹤注冊事件的句柄. 在 SelectionKey 對象的有效期間, Selector 會一直監(jiān)控與 SelectionKey 對象相關(guān)的事件, 如果事件發(fā)生, 就會把 SelectionKey 對象加入到 selected-keys 集合中. 在以下情況下, SelectionKey 對象會失效, 這意味著 Selector 再也不會監(jiān)控與它相關(guān)的事件了:

⑴ 程序調(diào)用 SelectionKey 的 cancel() 方法;

⑵ 關(guān)閉與 SelectionKey 關(guān)聯(lián)的 Channel;

⑶ 與 SelectionKey 關(guān)聯(lián)的 Selector 被關(guān)閉.

???? 在 SelectionKey 中定義了 4 種事件, 分別用 4 個 int 類型的常量來表示.

SelectionKey.OP_ACCEPT: 接收連接就緒事件, 表示服務器監(jiān)聽到了客戶連接, 服務器可以接收這個連接了. 常量值為 16.(00010000)
SelectionKey.OP_CONNECT: 連接就緒事件, 表示客戶與服務器的連接已經(jīng)建立成功. 常量值為 8.(00001000)
SelectionKey.OP_READ: 讀就緒事件, 表示通道中已經(jīng)有了可讀數(shù)據(jù), 可以執(zhí)行讀操作了. 常量值為 1.(00000001)
SelectionKey.OP_WRITE: 寫就緒事件, 表示已經(jīng)可以向通道寫數(shù)據(jù)了. 常量值為 4.(00000100)

????? 以上常量分別占據(jù)不同的二進制位, 因此可以通過二進制的或運算 "|", 來將它們進行任意組合. 一個 SelectionKey 對象中包含兩種類型的事件.

所有感興趣的事件: SelectionKey 的 interestOps() 方法返回所有感興趣的事件. 假如, 假定返回值為 SelectionKey.OP_WRITE | SelectionKey.OP_READ, 就表示這個 SelectionKey 對讀就緒和寫就緒事件感興趣. 與之關(guān)聯(lián)的 Selector 對象會負責監(jiān)控這些事件. 當通過 SelectableChannel 的 register() 方法注冊事件時, 可以在參數(shù)中指定 SelectionKey 感興趣的事件. 假如, 以下代碼表明新建的 SelectionKey 對連接就緒和讀就緒事件感興趣:

????????? SelectionKey key = socketChannel.register(selector,? SelectionKey.OP_CONNECT | SelectionKey.OP_READ );

???????SelectionKey 的 interestOps(int ops) 方法用于為 SelectionKey 對象增加一個感興趣的事件. 假如, 以下代碼使得 SelectionKey 增加了一個感興趣的事件:

??????????key.interestOps(?SelectionKey.OP_WRITE );?????????????????????????????????????????????????????

所有已經(jīng)發(fā)生的事件: SelectionKey 的 readyOps() 方法返回所有已經(jīng)發(fā)生的事件. 假如, 假定返回值為 SelectionKey.OP_WRITE | SelectionKey.OP_READ , 表示讀就緒和寫就緒事件發(fā)生了, 這意味著與之關(guān)聯(lián)的 SocketChannel 對象可以進行讀操作和寫操作了.

???? 當程序調(diào)用一個 SelectableChannel ()

總結(jié)

以上是生活随笔為你收集整理的非阻塞模式的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。

模式