elasticsearch入門系列">elasticsearch入門系列
813
2025-04-04
0x6 Java系列：Java NIO？看這一篇就夠了！【二】
四、TCP服務端的NIO寫法
到目前為止，所舉的案例中都沒有涉及Selector。不要急，好東西要慢慢來。Selector類可以用于避免使用阻塞式客戶端中很浪費資源的“忙等”方法。例如，考慮一個IM服務器。像QQ或者旺旺這樣的，可能有幾萬甚至幾千萬個客戶端同時連接到了服務器，但在任何時刻都只是非常少量的消息。
需要讀取和分發。這就需要一種方法阻塞等待，直到至少有一個信道可以進行I/O操作，并指出是哪個信道。NIO的選擇器就實現了這樣的功能。一個Selector實例可以同時檢查一組信道的I/O狀態。用專業術語來說，選擇器就是一個多路開關選擇器，因為一個選擇器能夠管理多個信道上的I/O操作。然而如果用傳統的方式來處理這么多客戶端，使用的方法是循環地一個一個地去檢查所有的客戶端是否有I/O操作，如果當前客戶端有I/O操作，則可能把當前客戶端扔給一個線程池去處理，如果沒有I/O操作則進行下一個輪詢，當所有的客戶端都輪詢過了又接著從頭開始輪詢；這種方法是非常笨而且也非常浪費資源，因為大部分客戶端是沒有I/O操作，我們也要去檢查；而Selector就不一樣了，它在內部可以同時管理多個I/O，當一個信道有I/O操作的時候，他會通知Selector，Selector就是記住這個信道有I/O操作，并且知道是何種I/O操作，是讀呢？是寫呢？還是接受新的連接；所以如果使用Selector，它返回的結果只有兩種結果，一種是0，即在你調用的時刻沒有任何客戶端需要I/O操作，另一種結果是一組需要I/O操作的客戶端，這時你就根本不需要再檢查了，因為它返回給你的肯定是你想要的。這樣一種通知的方式比那種主動輪詢的方式要高效得多！
要使用選擇器（Selector），需要創建一個Selector實例（使用靜態工廠方法open()）并將其注冊（register）到想要監控的信道上（注意，這要通過channel的方法實現，而不是使用selector的方法）。最后，調用選擇器的select()方法。該方法會阻塞等待，直到有一個或更多的信道準備好了I/O操作或等待超時。select()方法將返回可進行I/O操作的信道數量。現在，在一個單獨的線程中，通過調用select()方法就能檢查多個信道是否準備好進行I/O操作。如果經過一段時間后仍然沒有信道準備好，select()方法就會返回0，并允許程序繼續執行其他任務。
下面將上面的TCP服務端代碼改寫成NIO的方式（案例5）：
public class ServerConnect
{
private static final int BUF_SIZE=1024;
private static final int PORT = 8080;
private static final int TIMEOUT = 3000;
public static void main(String[] args)
{
selector();
}
public static void handleAccept(SelectionKey key) throws IOException{
ServerSocketChannel ssChannel = (ServerSocketChannel)key.channel();
SocketChannel sc = ssChannel.accept();
sc.configureBlocking(false);
sc.register(key.selector(), SelectionKey.OP_READ,ByteBuffer.allocateDirect(BUF_SIZE));
}
public static void handleRead(SelectionKey key) throws IOException{
SocketChannel sc = (SocketChannel)key.channel();
ByteBuffer buf = (ByteBuffer)key.attachment();
long bytesRead = sc.read(buf);
while(bytesRead>0){
buf.flip();
while(buf.hasRemaining()){
System.out.print((char)buf.get());
}
System.out.println();
buf.clear();
bytesRead = sc.read(buf);
}
if(bytesRead == -1){
sc.close();
}
}
public static void handleWrite(SelectionKey key) throws IOException{
ByteBuffer buf = (ByteBuffer)key.attachment();
buf.flip();
SocketChannel sc = (SocketChannel) key.channel();
while(buf.hasRemaining()){
sc.write(buf);
}
buf.compact();
}
public static void selector() {
Selector selector = null;
ServerSocketChannel ssc = null;
try{
selector = Selector.open();
ssc= ServerSocketChannel.open();
ssc.socket().bind(new InetSocketAddress(PORT));
ssc.configureBlocking(false);
ssc.register(selector, SelectionKey.OP_ACCEPT);
while(true){
if(selector.select(TIMEOUT) == 0){
System.out.println("==");
continue;
}
Iterator
while(iter.hasNext()){
SelectionKey key = iter.next();
if(key.isAcceptable()){
handleAccept(key);
}
if(key.isReadable()){
handleRead(key);
}
if(key.isWritable() && key.isValid()){
handleWrite(key);
}
if(key.isConnectable()){
System.out.println("isConnectable = true");
}
iter.remove();
}
}
}catch(IOException e){
e.printStackTrace();
}finally{
try{
if(selector!=null){
selector.close();
}
if(ssc!=null){
ssc.close();
}
}catch(IOException e){
e.printStackTrace();
}
}
}
}
下面來慢慢講解這段代碼。
ServerSocketChannel
打開ServerSocketChannel：
ServerSocketChannel?serverSocketChannel?=?ServerSocketChannel.open();
關閉ServerSocketChannel：
serverSocketChannel.close();
監聽新進來的連接：
while(true){
SocketChannel socketChannel = serverSocketChannel.accept();
}
ServerSocketChannel可以設置成非阻塞模式。在非阻塞模式下，accept()?方法會立刻返回，如果還沒有新進來的連接,返回的將是null。因此，需要檢查返回的SocketChannel是否是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...
}
}
Selector
Selector的創建：Selector selector = Selector.open();
為了將Channel和Selector配合使用，必須將Channel注冊到Selector上，通過SelectableChannel.register()方法來實現，沿用案例5中的部分代碼：
ssc= ServerSocketChannel.open();
ssc.socket().bind(new InetSocketAddress(PORT));
ssc.configureBlocking(false);
ssc.register(selector, SelectionKey.OP_ACCEPT);
與Selector一起使用時，Channel必須處于非阻塞模式下。這意味著不能將FileChannel與Selector一起使用，因為FileChannel不能切換到非阻塞模式。而套接字通道都可以。
注意register()方法的第二個參數。這是一個“interest集合”，意思是在通過Selector監聽Channel時對什么事件感興趣。可以監聽四種不同類型的事件：
1. Connect
2. Accept
3. Read
4. Write
通道觸發了一個事件意思是該事件已經就緒。所以，某個channel成功連接到另一個服務器稱為“連接就緒”。一個server socket channel準備好接收新進入的連接稱為“接收就緒”。一個有數據可讀的通道可以說是“讀就緒”。等待寫數據的通道可以說是“寫就緒”。
這四種事件用SelectionKey的四個常量來表示：
1. SelectionKey.OP_CONNECT
2. SelectionKey.OP_ACCEPT
3. SelectionKey.OP_READ
4. SelectionKey.OP_WRITE
SelectionKey
當向Selector注冊Channel時，register()方法會返回一個SelectionKey對象。這個對象包含了一些你感興趣的屬性：
·???interest集合
·???ready集合
·???Channel
·???Selector
·???附加的對象（可選）
interest集合：就像向Selector注冊通道一節中所描述的，interest集合是你所選擇的感興趣的事件集合。可以通過SelectionKey讀寫interest集合。
ready?集合是通道已經準備就緒的操作的集合。在一次選擇(Selection)之后，你會首先訪問這個ready set。Selection將在下一小節進行解釋。可以這樣訪問ready集合：
int?readySet?=?selectionKey.readyOps();
可以用像檢測interest集合那樣的方法，來檢測channel中什么事件或操作已經就緒。但是，也可以使用以下四個方法，它們都會返回一個布爾類型：
selectionKey.isAcceptable();
selectionKey.isConnectable();
selectionKey.isReadable();
selectionKey.isWritable();
從SelectionKey訪問Channel和Selector很簡單。如下：
Channel ?channel ?= selectionKey.channel();
Selector selector = selectionKey.selector();
可以將一個對象或者更多信息附著到SelectionKey上，這樣就能方便的識別某個給定的通道。例如，可以附加與通道一起使用的Buffer，或是包含聚集數據的某個對象。使用方法如下：
selectionKey.attach(theObject);
Object attachedObj = selectionKey.attachment();
還可以在用register()方法向Selector注冊Channel的時候附加對象。如：
SelectionKey?key?=?channel.register(selector,?SelectionKey.OP_READ,?theObject);
通過Selector選擇通道
一旦向Selector注冊了一或多個通道，就可以調用幾個重載的select()方法。這些方法返回你所感興趣的事件（如連接、接受、讀或寫）已經準備就緒的那些通道。換句話說，如果你對“讀就緒”的通道感興趣，select()方法會返回讀事件已經就緒的那些通道。
下面是select()方法：
·???int select()
·???int select(long timeout)
·???int selectNow()
select()阻塞到至少有一個通道在你注冊的事件上就緒了。
select(long timeout)和select()一樣，除了最長會阻塞timeout毫秒(參數)。
selectNow()不會阻塞，不管什么通道就緒都立刻返回（譯者注：此方法執行非阻塞的選擇操作。如果自從前一次選擇操作后，沒有通道變成可選擇的，則此方法直接返回零。）。
select()方法返回的int值表示有多少通道已經就緒。亦即，自上次調用select()方法后有多少通道變成就緒狀態。如果調用select()方法，因為有一個通道變成就緒狀態，返回了1，若再次調用select()方法，如果另一個通道就緒了，它會再次返回1。如果對第一個就緒的channel沒有做任何操作，現在就有兩個就緒的通道，但在每次select()方法調用之間，只有一個通道就緒了。
一旦調用了select()方法，并且返回值表明有一個或更多個通道就緒了，然后可以通過調用selector的selectedKeys()方法，訪問“已選擇鍵集（selected key set）”中的就緒通道。如下所示：
Set?selectedKeys?=?selector.selectedKeys();
當向Selector注冊Channel時，Channel.register()方法會返回一個SelectionKey?對象。這個對象代表了注冊到該Selector的通道。
注意每次迭代末尾的keyIterator.remove()調用。Selector不會自己從已選擇鍵集中移除SelectionKey實例。必須在處理完通道時自己移除。下次該通道變成就緒時，Selector會再次將其放入已選擇鍵集中。
SelectionKey.channel()方法返回的通道需要轉型成你要處理的類型，如ServerSocketChannel或SocketChannel等。
一個完整的使用Selector和ServerSocketChannel的案例可以參考案例5的selector()方法。
五、內存映射文件
JAVA處理大文件，一般用BufferedReader,BufferedInputStream這類帶緩沖的IO類，不過如果文件超大的話，更快的方式是采用MappedByteBuffer。
MappedByteBuffer是NIO引入的文件內存映射方案，讀寫性能極高。NIO最主要的就是實現了對異步操作的支持。其中一種通過把一個套接字通道(SocketChannel)注冊到一個選擇器(Selector)中,不時調用后者的選擇(select)方法就能返回滿足的選擇鍵(SelectionKey),鍵中包含了SOCKET事件信息。這就是select模型。
SocketChannel的讀寫是通過一個類叫ByteBuffer來操作的.這個類本身的設計是不錯的,比直接操作byte[]方便多了. ByteBuffer有兩種模式:直接/間接.間接模式最典型(也只有這么一種)的就是HeapByteBuffer,即操作堆內存(byte[]).但是內存畢竟有限,如果我要發送一個1G的文件怎么辦?不可能真的去分配1G的內存.這時就必須使用"直接"模式,即MappedByteBuffer,文件映射.
先中斷一下,談談操作系統的內存管理.一般操作系統的內存分兩部分:物理內存;虛擬內存.虛擬內存一般使用的是頁面映像文件,即硬盤中的某個(某些)特殊的文件.操作系統負責頁面文件內容的讀寫,這個過程叫"頁面中斷/切換". MappedByteBuffer也是類似的,你可以把整個文件(不管文件有多大)看成是一個ByteBuffer.MappedByteBuffer?只是一種特殊的ByteBuffer，即是ByteBuffer的子類。MappedByteBuffer?將文件直接映射到內存（這里的內存指的是虛擬內存，并不是物理內存）。通常，可以映射整個文件，如果文件比較大的話可以分段進行映射，只要指定文件的那個部分就可以。
概念
FileChannel提供了map方法來把文件影射為內存映像文件：MappedByteBuffer map(int mode,long position,long size);?可以把文件的從position開始的size大小的區域映射為內存映像文件，mode指出了可訪問該內存映像文件的方式：
·???READ_ONLY,（只讀）：試圖修改得到的緩沖區將導致拋出ReadOnlyBufferException.(MapMode.READ_ONLY)
·???READ_WRITE（讀/寫）：對得到的緩沖區的更改最終將傳播到文件；該更改對映射到同一文件的其他程序不一定是可見的。(MapMode.READ_WRITE)
·???PRIVATE（專用）：對得到的緩沖區的更改不會傳播到文件，并且該更改對映射到同一文件的其他程序也不是可見的；相反，會創建緩沖區已修改部分的專用副本。(MapMode.PRIVATE)
MappedByteBuffer是ByteBuffer的子類，其擴充了三個方法：
·???force()：緩沖區是READ_WRITE模式下，此方法對緩沖區內容的修改強行寫入文件；
·???load()：將緩沖區的內容載入內存，并返回該緩沖區的引用；
·???isLoaded()：如果緩沖區的內容在物理內存中，則返回真，否則返回假；
案例對比
這里通過采用ByteBuffer和MappedByteBuffer分別讀取大小約為5M的文件"src/1.ppt"來比較兩者之間的區別，method3()是采用MappedByteBuffer讀取的，method4()對應的是ByteBuffer。
public static void method4(){
RandomAccessFile aFile = null;
FileChannel fc = null;
try{
aFile = new RandomAccessFile("src/1.ppt","rw");
fc = aFile.getChannel();
long timeBegin = System.currentTimeMillis();
ByteBuffer buff = ByteBuffer.allocate((int) aFile.length());
buff.clear();
fc.read(buff);
//System.out.println((char)buff.get((int)(aFile.length()/2-1)));
//System.out.println((char)buff.get((int)(aFile.length()/2)));
//System.out.println((char)buff.get((int)(aFile.length()/2)+1));
long timeEnd = System.currentTimeMillis();
System.out.println("Read time: "+(timeEnd-timeBegin)+"ms");
}catch(IOException e){
e.printStackTrace();
}finally{
try{
if(aFile!=null){
aFile.close();
}
if(fc!=null){
fc.close();
}
}catch(IOException e){
e.printStackTrace();
}
}
}
public static void method3(){
RandomAccessFile aFile = null;
FileChannel fc = null;
try{
aFile = new RandomAccessFile("src/1.ppt","rw");
fc = aFile.getChannel();
long timeBegin = System.currentTimeMillis();
MappedByteBuffer mbb = fc.map(FileChannel.MapMode.READ_ONLY, 0, aFile.length());
// System.out.println((char)mbb.get((int)(aFile.length()/2-1)));
// System.out.println((char)mbb.get((int)(aFile.length()/2)));
//System.out.println((char)mbb.get((int)(aFile.length()/2)+1));
long timeEnd = System.currentTimeMillis();
System.out.println("Read time: "+(timeEnd-timeBegin)+"ms");
}catch(IOException e){
e.printStackTrace();
}finally{
try{
if(aFile!=null){
aFile.close();
}
if(fc!=null){
fc.close();
}
}catch(IOException e){
e.printStackTrace();
}
}
}
通過在入口函數main()中運行：
method3();
System.out.println("=============");
method4();
輸出結果（運行在普通PC機上）：
Read time: 2ms
=============
Read time: 12ms
通過輸出結果可以看出彼此的差別，一個例子也許是偶然，那么下面把5M大小的文件替換為200M的文件，輸出結果：
Read time: 1ms
=============
Read time: 407ms
可以看到差距拉大。
注：MappedByteBuffer有資源釋放的問題：被MappedByteBuffer打開的文件只有在垃圾收集時才會被關閉，而這個點是不確定的。在Javadoc中這里描述：A mapped byte buffer and the file mapping that it represents remian valid until the buffer itself is garbage-collected。詳細可以翻閱參考資料5和6.
六、其余功能介紹
看完以上陳述，詳細大家對NIO有了一定的了解，下面主要通過幾個案例，來說明NIO的其余功能，下面代碼量偏多，功能性講述偏少。
Scatter/Gatter
分散（scatter）從Channel中讀取是指在讀操作時將讀取的數據寫入多個buffer中。因此，Channel將從Channel中讀取的數據“分散（scatter）”到多個Buffer中。
聚集（gather）寫入Channel是指在寫操作時將多個buffer的數據寫入同一個Channel，因此，Channel?將多個Buffer中的數據“聚集（gather）”后發送到Channel。
scatter / gather經常用于需要將傳輸的數據分開處理的場合，例如傳輸一個由消息頭和消息體組成的消息，你可能會將消息體和消息頭分散到不同的buffer中，這樣你可以方便的處理消息頭和消息體。
案例：
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.nio.ByteBuffer;
import java.nio.channels.Channel;
import java.nio.channels.FileChannel;
public class ScattingAndGather
{
public static void main(String args[]){
gather();
}
public static void gather()
{
ByteBuffer header = ByteBuffer.allocate(10);
ByteBuffer body = ByteBuffer.allocate(10);
byte [] b1 = {'0', '1'};
byte [] b2 = {'2', '3'};
header.put(b1);
body.put(b2);
ByteBuffer [] buffs = {header, body};
try
{
FileOutputStream os = new FileOutputStream("src/scattingAndGather.txt");
FileChannel channel = os.getChannel();
channel.write(buffs);
}
catch (IOException e)
{
e.printStackTrace();
}
}
}
transferFrom & transferTo
FileChannel的transferFrom()方法可以將數據從源通道傳輸到FileChannel中。
public static void method1(){
RandomAccessFile fromFile = null;
RandomAccessFile toFile = null;
try
{
fromFile = new RandomAccessFile("src/fromFile.xml","rw");
FileChannel fromChannel = fromFile.getChannel();
toFile = new RandomAccessFile("src/toFile.txt","rw");
FileChannel toChannel = toFile.getChannel();
long position = 0;
long count = fromChannel.size();
System.out.println(count);
toChannel.transferFrom(fromChannel, position, count);
}
catch (IOException e)
{
e.printStackTrace();
}
finally{
try{
if(fromFile != null){
fromFile.close();
}
if(toFile != null){
toFile.close();
}
}
catch(IOException e){
e.printStackTrace();
}
}
}
方法的輸入參數position表示從position處開始向目標文件寫入數據，count表示最多傳輸的字節數。如果源通道的剩余空間小于count?個字節，則所傳輸的字節數要小于請求的字節數。此外要注意，在SoketChannel的實現中，SocketChannel只會傳輸此刻準備好的數據（可能不足count字節）。因此，SocketChannel可能不會將請求的所有數據(count個字節)全部傳輸到FileChannel中。
transferTo()方法將數據從FileChannel傳輸到其他的channel中。
public static void method2()
{
RandomAccessFile fromFile = null;
RandomAccessFile toFile = null;
try
{
fromFile = new RandomAccessFile("src/fromFile.txt","rw");
FileChannel fromChannel = fromFile.getChannel();
toFile = new RandomAccessFile("src/toFile.txt","rw");
FileChannel toChannel = toFile.getChannel();
long position = 0;
long count = fromChannel.size();
System.out.println(count);
fromChannel.transferTo(position, count,toChannel);
}
catch (IOException e)
{
e.printStackTrace();
}
finally{
try{
if(fromFile != null){
fromFile.close();
}
if(toFile != null){
toFile.close();
}
}
catch(IOException e){
e.printStackTrace();
}
}
}
上面所說的關于SocketChannel的問題在transferTo()方法中同樣存在。SocketChannel會一直傳輸數據直到目標buffer被填滿。
Pipe
Java NIO?管道是2個線程之間的單向數據連接。Pipe有一個source通道和一個sink通道。數據會被寫到sink通道，從source通道讀取。
public static void method1(){
Pipe pipe = null;
ExecutorService exec = Executors.newFixedThreadPool(2);
try{
pipe = Pipe.open();
final Pipe pipeTemp = pipe;
exec.submit(new Callable
@Override
public Object call() throws Exception
{
Pipe.SinkChannel sinkChannel = pipeTemp.sink();//向通道中寫數據
while(true){
TimeUnit.SECONDS.sleep(1);
String newData = "Pipe Test At Time "+System.currentTimeMillis();
ByteBuffer buf = ByteBuffer.allocate(1024);
buf.clear();
buf.put(newData.getBytes());
buf.flip();
while(buf.hasRemaining()){
System.out.println(buf);
sinkChannel.write(buf);
}
}
}
});
exec.submit(new Callable
@Override
public Object call() throws Exception
{
Pipe.SourceChannel sourceChannel = pipeTemp.source();//向通道中讀數據
while(true){
TimeUnit.SECONDS.sleep(1);
ByteBuffer buf = ByteBuffer.allocate(1024);
buf.clear();
int bytesRead = sourceChannel.read(buf);
System.out.println("bytesRead="+bytesRead);
while(bytesRead >0 ){
buf.flip();
byte b[] = new byte[bytesRead];
int i=0;
while(buf.hasRemaining()){
b[i]=buf.get();
System.out.printf("%X",b[i]);
i++;
}
String s = new String(b);
System.out.println("=================||"+s);
bytesRead = sourceChannel.read(buf);
}
}
}
});
}catch(IOException e){
e.printStackTrace();
}finally{
exec.shutdown();
}
}
DatagramChannel
Java NIO中的DatagramChannel是一個能收發UDP包的通道。因為UDP是無連接的網絡協議，所以不能像其它通道那樣讀取和寫入。它發送和接收的是數據包。
public static void ?reveive(){
DatagramChannel channel = null;
try{
channel = DatagramChannel.open();
channel.socket().bind(new InetSocketAddress(8888));
ByteBuffer buf = ByteBuffer.allocate(1024);
buf.clear();
channel.receive(buf);
buf.flip();
while(buf.hasRemaining()){
System.out.print((char)buf.get());
}
System.out.println();
}catch(IOException e){
e.printStackTrace();
}finally{
try{
if(channel!=null){
channel.close();
}
}catch(IOException e){
e.printStackTrace();
}
}
}
public static void send(){
DatagramChannel channel = null;
try{
channel = DatagramChannel.open();
String info = "I'm the Sender!";
ByteBuffer buf = ByteBuffer.allocate(1024);
buf.clear();
buf.put(info.getBytes());
buf.flip();
int bytesSent = channel.send(buf, new InetSocketAddress("10.10.195.115",8888));
System.out.println(bytesSent);
}catch(IOException e){
e.printStackTrace();
}finally{
try{
if(channel!=null){
channel.close();
}
}catch(IOException e){
e.printStackTrace();
}
}
}
版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。
版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。
相關文章
