NIO概述
Java NIO全稱為Non-blocking IO或者New IO顿仇,從名字我們知道NIO是非阻塞的IO笤闯,而Java IO則是阻塞的IO差凹。在一般的情況下阻塞是低效率的老赤,特別是在高并發(fā)的場景下面逸嘀,因此Java引入了NIO。NIO相比IO來說主要有以下幾個區(qū)別:
- NIO是面向緩沖區(qū)的碰纬,IO則面向流萍聊。
- 標準的IO編程接口是面向字節(jié)流和字符流的。而NIO是面向通道和緩沖區(qū)的悦析,數(shù)據(jù)總是從通道中讀到buffer緩沖區(qū)內(nèi)寿桨,或者從buffer緩沖區(qū)寫入到通道中;( NIO中的所有I/O操作都是通過一個通道開始的她按。)
- Java IO面向流意味著每次從流中讀一個或多個字節(jié)牛隅,直至讀取所有字節(jié),它們沒有被緩存在任何地方酌泰;
- Java NIO是面向緩存的I/O方法媒佣。 將數(shù)據(jù)讀入緩沖器,使用通道進一步處理數(shù)據(jù)陵刹。 在NIO中默伍,使用通道和緩沖區(qū)來處理I/O操作。
- NIO是非阻塞的衰琐,IO是阻塞的也糊。
- Java NIO使我們可以進行非阻塞IO操作。比如說羡宙,單線程中從通道讀取數(shù)據(jù)到buffer狸剃,同時可以繼續(xù)做別的事情,當數(shù)據(jù)讀取到buffer中后狗热,線程再繼續(xù)處理數(shù)據(jù)钞馁。寫數(shù)據(jù)也是一樣的。另外匿刮,非阻塞寫也是如此僧凰。一個線程請求寫入一些數(shù)據(jù)到某通道,但不需要等待它完全寫入熟丸,這個線程同時可以去做別的事情训措。
- Java IO的各種流是阻塞的。這意味著,當一個線程調(diào)用read() 或 write()時绩鸣,該線程被阻塞怀大,直到有一些數(shù)據(jù)被讀取,或數(shù)據(jù)完全寫入呀闻。該線程在此期間不能再干任何事情了
- NIO有Selectors(多路復用器)叉寂,而IO沒有Selectors。
- 選擇器用于使用單個線程處理多個通道总珠。因此,它需要較少的線程來處理這些通道勘纯。
- 線程之間的切換對于操作系統(tǒng)來說是昂貴的局服。 因此,為了提高系統(tǒng)效率選擇器是有用的
NIO中主要有以下三個概念:通道驳遵、緩沖區(qū)和Selectors淫奔。
通道
Java NIO Channel通道和流非常相似,主要有以下幾點區(qū)別:
- 通道可以讀也可以寫堤结,流一般來說是單向的(只能讀或者寫)唆迁。
- 通道可以異步讀寫。
- 通道總是基于緩沖區(qū)Buffer來讀寫竞穷。
Java的NIO讀寫都是在通道中進行的唐责,通道涵蓋了網(wǎng)絡UDP,TCP網(wǎng)絡IO和文件IO:
- DatagramChannel
- SocketChannel
- FileChannel
- ServerSocketChannel
各個Channel的UML類圖如下:
DatagramChannel
DatagramChannel用于處理UDP連接瘾带。
打開一個DatagramChannel
DatagramChannel channel = DatagramChannel.open();
channel.socket().bind(new InetSocketAddress(8888));
讀取數(shù)據(jù)
ByteBuffer buf = ByteBuffer.allocate(1024);
buf.clear();
channel.receive(buf);
發(fā)送數(shù)據(jù)
String msg = "Current time is: " + System.currentTimeMillis();
ByteBuffer buf = ByteBuffer.allocate(1024);
buf.clear();
buf.put(msg.getBytes());
buf.flip();
int bytesSent = channel.send(buf, new InetSocketAddress("host", port));
SocketChannel
打開 SocketChannel
SocketChannel socketChannel = SocketChannel.open();
socketChannel.connect(new InetSocketAddress("host", 80));
讀取數(shù)據(jù)
ByteBuffer buf = ByteBuffer.allocate(48);
int bytesRead = socketChannel.read(buf);
如果 read()返回 -1, 表明連接已經(jīng)中斷鼠哥。
寫入數(shù)據(jù)
String msg = "Current Time is: " + System.currentTimeMillis();
ByteBuffer buf = ByteBuffer.allocate(48);
buf.clear();
buf.put(msg.getBytes());
buf.flip();
while(buf.hasRemaining()) {
channel.write(buf);
}
非阻塞模式
socketChannel.configureBlocking(false);
socketChannel.connect(new InetSocketAddress("host", 80));
while (!socketChannel.finishConnect()) {
}
我們可以設置 SocketChannel 為異步模式, 這樣 connect, read, write 都是異步的了。在異步模式中, 或許連接還沒有建立, connect 方法就返回了, 因此我們需要檢查當前是否是連接到了主機看政,因此通過一個 while 循環(huán)來判斷朴恳。
FileChannel
打開
RandomAccessFile aFile = new RandomAccessFile("test.txt", "rw");
FileChannel inChannel = aFile.getChannel();
讀取
ByteBuffer buf = ByteBuffer.allocate(48);
int bytesRead = inChannel.read(buf);
寫入
String newData = "Current time is: " + System.currentTimeMillis();
ByteBuffer buf = ByteBuffer.allocate(1024);
buf.clear();
buf.put(newData.getBytes());
buf.flip();
while (buf.hasRemaining()) {
channel.write(buf);
}
關(guān)閉
channel.close();
設置 position
long pos = channel.position();
channel.position(pos + 123);
文件大小
我們可以通過 channel.size()獲取關(guān)聯(lián)到這個 Channel 中的文件的大小。注意, 這里返回的是文件的大小, 而不是 Channel 中剩余的元素個數(shù)允蚣。
截斷文件
channel.truncate(1024);
將文件的大小截斷為1024字節(jié)于颖。
強制寫入
channel.force(true);
強制將緩存中的數(shù)據(jù)寫入文件中:
ServerSocketChannel
ServerSocketChannel顧名思義,它是用來監(jiān)聽server端的socket連接嚷兔。
打開和關(guān)閉
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.close();
監(jiān)聽連接
使用ServerSocketChannel的accept()方法來監(jiān)聽客戶端的TCP連接請求森渐,accept()方法是阻塞的,直到有連接進來谴垫。
while(true){
SocketChannel socketChannel = serverSocketChannel.accept();
//do something with socketChannel...
}
非阻塞模式
如果設定ServerSocketChannel是非阻塞的章母,則accept()方法不會阻塞。如果返回的是null證明沒有新的連接翩剪,如果不是null乳怎,則有新的連接請求。
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.socket().bind(new InetSocketAddress(9999));
serverSocketChannel.configureBlocking(false);
while (true) {
SocketChannel socketChannel = serverSocketChannel.accept();
if(socketChannel != null) {
// do something with socketChannel...
}
}
Buffer緩沖區(qū)
Java NIO Buffers用于和NIO Channel交互。正如你已經(jīng)知道的蚪缀,我們從channel中讀取數(shù)據(jù)到buffers里秫逝,從buffer把數(shù)據(jù)寫入到channels。
buffer本質(zhì)上就是一塊內(nèi)存區(qū)询枚,可以用來寫入數(shù)據(jù)违帆,并在稍后讀取出來。這塊內(nèi)存被NIO Buffer包裹起來金蜀,對外提供一系列的讀寫方便開發(fā)的接口刷后。
Buffer基本用法
利用Buffer讀寫數(shù)據(jù),通常遵循四個步驟:
- 把數(shù)據(jù)寫入buffer渊抄;
- 調(diào)用flip尝胆;
- 從Buffer中讀取數(shù)據(jù);
- 調(diào)用buffer.clear()或者buffer.compact()
當寫入數(shù)據(jù)到buffer中時护桦,buffer會記錄已經(jīng)寫入的數(shù)據(jù)大小含衔。當需要讀數(shù)據(jù)時,通過flip()方法把buffer從寫模式調(diào)整為讀模式二庵;在讀模式下贪染,可以讀取所有已經(jīng)寫入的數(shù)據(jù)。
當讀取完數(shù)據(jù)后催享,需要清空buffer杭隙,以滿足后續(xù)寫入操作。清空buffer有兩種方式:調(diào)用clear()或compact()方法因妙。clear會清空整個buffer寺渗,compact則只清空已讀取的數(shù)據(jù),未被讀取的數(shù)據(jù)會被移動到buffer的開始位置兰迫,寫入位置則近跟著未讀數(shù)據(jù)之后信殊。
這里有一個簡單的buffer案例,包括了write汁果,flip和clear操作:
RandomAccessFile aFile = new RandomAccessFile("data/nio-data.txt", "rw");
FileChannel inChannel = aFile.getChannel();
// create buffer with capacity of 48 bytes
ByteBuffer buf = ByteBuffer.allocate(48);
int bytesRead = inChannel.read(buf); //read into buffer.
while (bytesRead != -1) {
buf.flip(); // make buffer ready for read
while(buf.hasRemaining()){
System.out.print((char) buf.get()); // read 1 byte at a time
}
buf.clear(); //make buffer ready for writing
bytesRead = inChannel.read(buf);
}
aFile.close();
Buffer的容量涡拘,位置,上限(Buffer Capacity, Position and Limit)
buffer緩沖區(qū)實質(zhì)上就是一塊內(nèi)存据德,用于寫入數(shù)據(jù)鳄乏,也供后續(xù)再次讀取數(shù)據(jù)。這塊內(nèi)存被NIO Buffer管理棘利,并提供一系列的方法用于更簡單的操作這塊內(nèi)存橱野。
一個Buffer有三個屬性是必須掌握的,分別是:
- capacity容量
- position位置
- limit限制
position和limit的具體含義取決于當前buffer的模式善玫。capacity在兩種模式下都表示容量水援。
下面有張示例圖,描訴了不同模式下position和limit的含義:
容量(Capacity)
作為一塊內(nèi)存,buffer有一個固定的大小蜗元,叫做capacity容量或渤。也就是最多只能寫入容量值得字節(jié),整形等數(shù)據(jù)奕扣。一旦buffer寫滿了就需要清空已讀數(shù)據(jù)以便下次繼續(xù)寫入新的數(shù)據(jù)薪鹦。
位置(Position)
當寫入數(shù)據(jù)到Buffer的時候需要中一個確定的位置開始,默認初始化時這個位置position為0惯豆,一旦寫入了數(shù)據(jù)比如一個字節(jié)池磁,整形數(shù)據(jù),那么position的值就會指向數(shù)據(jù)之后的一個單元楷兽,position最大可以到capacity - 1框仔。
當從Buffer讀取數(shù)據(jù)時,也需要從一個確定的位置開始拄养。buffer從寫入模式變?yōu)樽x取模式時,position會歸零银舱,每次讀取后瘪匿,position向后移動。
上限(Limit)
在寫模式寻馏,limit的含義是我們所能寫入的最大數(shù)據(jù)量棋弥。它等同于buffer的容量。
一旦切換到讀模式诚欠,limit則代表我們所能讀取的最大數(shù)據(jù)量顽染,他的值等同于寫模式下position的位置。
數(shù)據(jù)讀取的上限時buffer中已有的數(shù)據(jù)轰绵,也就是limit的位置(原position所指的位置)粉寞。
Buffer Types
Java NIO有如下具體的Buffer類型:
ByteBuffer
MappedByteBuffer
CharBuffer
DoubleBuffer
FloatBuffer
IntBuffer
LongBuffer
ShortBuffer
正如你看到的,Buffer的類型代表了不同數(shù)據(jù)類型左腔,換句話說唧垦,Buffer中的數(shù)據(jù)可以是上述的基本類型;
分配一個Buffer(Allocating a Buffer)
為了獲取一個Buffer對象液样,你必須先分配振亮。每個Buffer實現(xiàn)類都有一個allocate()方法用于分配內(nèi)存。下面看一個實例,開辟一個48字節(jié)大小的buffer:
ByteBuffer buf = ByteBuffer.allocate(48);
開辟一個1024個字符的CharBuffer:
CharBuffer buf = CharBuffer.allocate(1024);
寫入數(shù)據(jù)到Buffer(Writing Data to a Buffer)
寫數(shù)據(jù)到Buffer有兩種方法:
- 從Channel中寫數(shù)據(jù)到Buffer鞭莽。
- 手動寫數(shù)據(jù)到Buffer坊秸,調(diào)用put方法。
下面是一個實例澎怒,演示從Channel寫數(shù)據(jù)到Buffer:
int bytesRead = inChannel.read(buf); // read into buffer
通過put寫數(shù)據(jù):
buf.put(127);
put方法有很多不同版本褒搔,對應不同的寫數(shù)據(jù)方法。例如把數(shù)據(jù)寫到特定的位置,或者把一個字節(jié)數(shù)據(jù)寫入buffer站超≥┧。看考JavaDoc文檔可以查閱的更多數(shù)據(jù)。
翻轉(zhuǎn)(flip())
flip()方法可以吧Buffer從寫模式切換到讀模式死相。調(diào)用flip方法會把position歸零融求,并設置limit為之前的position的值。也就是說算撮,現(xiàn)在position代表的是讀取位置生宛,limit表示的是已寫入的數(shù)據(jù)位置。
從Buffer讀取數(shù)據(jù)(Reading Data from a Buffer)
沖Buffer讀數(shù)據(jù)也有兩種方式肮柜。
- 從buffer讀數(shù)據(jù)到channel
- 從buffer直接讀取數(shù)據(jù)陷舅,調(diào)用get方法
讀取數(shù)據(jù)到channel的例子:
// read from buffer into channel.
int bytesWritten = inChannel.write(buf);
調(diào)用get讀取數(shù)據(jù)的例子:
byte aByte = buf.get();
get也有諸多版本,對應了不同的讀取方式审洞。
rewind()
Buffer.rewind()方法將position置為0莱睁,這樣我們可以重復讀取buffer中的數(shù)據(jù)。limit保持不變芒澜。
clear() and compact()
一旦我們從buffer中讀取完數(shù)據(jù)仰剿,需要復用buffer為下次寫數(shù)據(jù)做準備。只需要調(diào)用clear或compact方法痴晦。
clear方法會重置position為0南吮,limit為capacity,也就是整個Buffer清空誊酌。實際上Buffer中數(shù)據(jù)并沒有清空部凑,我們只是把標記為修改了。
如果Buffer還有一些數(shù)據(jù)沒有讀取完碧浊,調(diào)用clear就會導致這部分數(shù)據(jù)被“遺忘”涂邀,因為我們沒有標記這部分數(shù)據(jù)未讀。
針對這種情況箱锐,如果需要保留未讀數(shù)據(jù)必孤,那么可以使用compact。 因此compact和clear的區(qū)別就在于對未讀數(shù)據(jù)的處理瑞躺,是保留這部分數(shù)據(jù)還是一起清空敷搪。
mark() and reset()
通過mark方法可以標記當前的position,通過reset來恢復mark的位置幢哨,這個非常像canvas的save和restore:
buffer.mark();
// call buffer.get() a couple of times, e.g. during parsing.
buffer.reset(); // set position back to mark.
Selector
Selector是Java NIO中的一個組件赡勘,用于檢查一個或多個NIO Channel的狀態(tài)是否處于可讀、可寫捞镰。如此可以實現(xiàn)單線程管理多個channels,也就是可以管理多個網(wǎng)絡鏈接闸与。
為什么使用Selector(Why Use a Selector?)
用單線程處理多個channels的好處是我需要更少的線程來處理channel毙替。實際上,你甚至可以用一個線程來處理所有的channels践樱。從操作系統(tǒng)的角度來看厂画,切換線程開銷是比較昂貴的,并且每個線程都需要占用系統(tǒng)資源拷邢,因此暫用線程越少越好袱院。
需要留意的是,現(xiàn)代操作系統(tǒng)和CPU在多任務處理上已經(jīng)變得越來越好瞭稼,所以多線程帶來的影響也越來越小忽洛。如果一個CPU是多核的,如果不執(zhí)行多任務反而是浪費了機器的性能环肘。不過這些設計討論是另外的話題了欲虚。簡而言之,通過Selector我們可以實現(xiàn)單線程操作多個channel悔雹。
這有一幅示意圖复哆,描述了單線程處理三個channel的情況:
創(chuàng)建Selector(Creating a Selector)
創(chuàng)建一個Selector可以通過Selector.open()方法:
Selector selector = Selector.open();
注冊Channel到Selector上(Registering Channels with the Selector)
為了同Selector掛了Channel,我們必須先把Channel注冊到Selector上腌零,這個操作使用SelectableChannel.register():
channel.configureBlocking(false);
SelectionKey key = channel.register(selector, SelectionKey.OP_READ);
Channel必須是非阻塞的梯找。所以FileChannel不適用Selector,因為FileChannel不能切換為非阻塞模式莱没。Socket channel可以正常使用。
注意register的第二個參數(shù)酷鸦,這個參數(shù)是一個“關(guān)注集合”饰躲,代表我們關(guān)注的channel狀態(tài),有四種基礎類型可供監(jiān)聽:
- Connect
- Accept
- Read
- Write
一個channel觸發(fā)了一個事件也可視作該事件處于就緒狀態(tài)臼隔。因此當channel與server連接成功后嘹裂,那么就是“連接就緒”狀態(tài)。server channel接收請求連接時處于“可連接就緒”狀態(tài)摔握。channel有數(shù)據(jù)可讀時處于“讀就緒”狀態(tài)寄狼。channel可以進行數(shù)據(jù)寫入時處于“寫就緒”狀態(tài)。
上述的四種就緒狀態(tài)用SelectionKey中的常量表示如下:
- SelectionKey.OP_CONNECT
- SelectionKey.OP_ACCEPT
- SelectionKey.OP_READ
- SelectionKey.OP_WRITE
如果對多個事件感興趣可利用位的或運算結(jié)合多個常量氨淌,比如:
int interestSet = SelectionKey.OP_READ | SelectionKey.OP_WRITE;
SelectionKey's
在上一小節(jié)中泊愧,我們利用register方法把Channel注冊到了Selectors上,這個方法的返回值是SelectionKeys盛正,這個返回的對象包含了一些比較有價值的屬性:
- The interest set
- The ready set
- The Channel
- The Selector
- An attached object (optional)
Interest Set
這個“關(guān)注集合”實際上就是我們希望處理的事件的集合删咱,它的值就是注冊時傳入的參數(shù),我們可以用按為與運算把每個事件取出來:
int interestSet = selectionKey.interestOps();
boolean isInterestedInAccept = interestSet & SelectionKey.OP_ACCEPT;
boolean isInterestedInConnect = interestSet & SelectionKey.OP_CONNECT;
boolean isInterestedInRead = interestSet & SelectionKey.OP_READ;
boolean isInterestedInWrite = interestSet & SelectionKey.OP_WRITE;
Ready Set
"就緒集合"中的值是當前channel處于就緒的值豪筝,一般來說在調(diào)用了select方法后都會需要用到就緒狀態(tài)痰滋,select的介紹在胡須文章中繼續(xù)展開摘能。
int readySet = selectionKey.readyOps();
從“就緒集合”中取值的操作類似于“關(guān)注集合”的操作,當然還有更簡單的方法敲街,SelectionKey提供了一系列返回值為boolean的的方法:
selectionKey.isAcceptable();
selectionKey.isConnectable();
selectionKey.isReadable();
selectionKey.isWritable();
Channel + Selector
從SelectionKey操作Channel和Selector非常簡單:
Channel channel = selectionKey.channel();
Selector selector = selectionKey.selector();
Attaching Objects
我們可以給一個SelectionKey附加一個Object,這樣做一方面可以方便我們識別某個特定的channel,同時也增加了channel相關(guān)的附加信息亭病。例如因悲,可以把用于channel的buffer附加到SelectionKey上:
selectionKey.attach(theObject);
Object attachedObj = selectionKey.attachment();
附加對象的操作也可以在register的時候就執(zhí)行:
SelectionKey key = channel.register(selector, SelectionKey.OP_READ, theObject);
從Selector中選擇channel(Selecting Channels via a Selector)
一旦我們向Selector注冊了一個或多個channel后,就可以調(diào)用select來獲取channel墩蔓。select方法會返回所有處于就緒狀態(tài)的channel梢莽。 select方法具體如下:
- int select()
- int select(long timeout)
- int selectNow()
select()方法在返回channel之前處于阻塞狀態(tài)。 select(long timeout)和select做的事一樣奸披,不過他的阻塞有一個超時限制昏名。
selectNow()不會阻塞,根據(jù)當前狀態(tài)立刻返回合適的channel阵面。
select()方法的返回值是一個int整形轻局,代表有多少channel處于就緒了。也就是自上一次select后有多少channel進入就緒样刷。舉例來說仑扑,假設第一次調(diào)用select時正好有一個channel就緒,那么返回值是1置鼻,并且對這個channel做任何處理镇饮,接著再次調(diào)用select,此時恰好又有一個新的channel就緒箕母,那么返回值還是1储藐,現(xiàn)在我們一共有兩個channel處于就緒,但是在每次調(diào)用select時只有一個channel是就緒的嘶是。
selectedKeys()
在調(diào)用select并返回了有channel就緒之后钙勃,可以通過選中的key集合來獲取channel,這個操作通過調(diào)用selectedKeys()方法:
Set<SelectionKey> selectedKeys = selector.selectedKeys();
還記得在register時的操作吧聂喇,我們register后的返回值就是SelectionKey實例辖源,也就是我們現(xiàn)在通過selectedKeys()方法所返回的SelectionKey。
遍歷這些SelectionKey可以通過如下方法:
Set<SelectionKey> selectedKeys = selector.selectedKeys();
Iterator<SelectionKey> keyIterator = selectedKeys.iterator();
while(keyIterator.hasNext()) {
SelectionKey key = keyIterator.next();
if(key.isAcceptable()) {
// a connection was accepted by a ServerSocketChannel.
} else if (key.isConnectable()) {
// a connection was established with a remote server.
} else if (key.isReadable()) {
// a channel is ready for reading
} else if (key.isWritable()) {
// a channel is ready for writing
}
keyIterator.remove();
}
上述循環(huán)會迭代key集合希太,針對每個key我們單獨判斷他是處于何種就緒狀態(tài)克饶。
注意keyIterator.remove()方法的調(diào)用,Selector本身并不會移除SelectionKey對象誊辉,這個操作需要我們收到執(zhí)行彤路。當下次channel處于就緒是,Selector任然會吧這些key再次加入進來芥映。
SelectionKey.channel返回的channel實例需要強轉(zhuǎn)為我們實際使用的具體的channel類型洲尊,例如ServerSocketChannel或SocketChannel.
wakeUp()
由于調(diào)用select而被阻塞的線程远豺,可以通過調(diào)用Selector.wakeup()來喚醒即便此時已然沒有channel處于就緒狀態(tài)。具體操作是坞嘀,在另外一個線程調(diào)用wakeup躯护,被阻塞與select方法的線程就會立刻返回。
close()
當操作Selector完畢后丽涩,需要調(diào)用close方法棺滞。close的調(diào)用會關(guān)閉Selector并使相關(guān)的SelectionKey都無效。channel本身不管被關(guān)閉矢渊。
完整的Selector案例
這有一個完整的案例继准,首先打開一個Selector,然后注冊channel,最后檢測Selector的狀態(tài):
Selector selector = Selector.open();
channel.configureBlocking(false);
SelectionKey key = channel.register(selector, SelectionKey.OP_READ);
while (true) {
int readyChannels = selector.select();
if (readyChannels == 0) continue;
Set<SelectionKey> selectedKeys = selector.selectedKeys();
Iterator<SelectionKey> keyIterator = selectedKeys.iterator();
while (keyIterator.hasNext()) {
SelectionKey key = keyIterator.next();
if (key.isAcceptable()) {
// a connection was accepted by a ServerSocketChannel.
} else if (key.isConnectable()) {
// a connection was established with a remote server.
} else if (key.isReadable()) {
// a channel is ready for reading
} else if (key.isWritable()) {
// a channel is ready for writing
}
keyIterator.remove();
}
}