Java NIO及其常用API
一、初識NIO
> 在 JDK 1. 4 中 新 加入 了 NIO( New Input/ Output) 類, 引入了一種基於通道和緩衝區的 I/O 方式,它可以使用 Native 函式庫直接分配堆外記憶體,然後通過一個儲存在 Java 堆的 DirectByteBuffer 物件作為這塊記憶體的引用進行操作,避免了在 Java 堆和 Native 堆中來回複製資料。NIO 是一種同步非阻塞的 IO 模型。同步是指執行緒不斷輪詢 IO 事件是否就緒,非阻塞是指執行緒在等待 IO 的時候,可以同時做其他任務。同步的核心就是 Selector,Selector 代替了執行緒本身輪詢 IO 事件,避免了阻塞同時減少了不必要的執行緒消耗;非阻塞的核心就是通道和緩衝區,當 IO 事件就緒時,可以通過寫道緩衝區,保證 IO 的成功,而無需執行緒阻塞式地等待。 在NIO中有幾個比較關鍵的概念:Buffer(緩衝區),Channel(通道),Selector(選擇器)。
Buffer(緩衝區):
在Java NIO中負責資料的存取,緩衝區就是陣列,用於儲存不同型別的緩衝區。為什麼說NIO是基於緩衝區的IO方式呢?因為,當一個連結建立完成後,IO的資料未必會馬上到達,為了當資料到達時能夠正確完成IO操作,在BIO(阻塞IO)中,等待IO的執行緒必須被阻塞,以全天候地執行IO操作。為了解決這種IO方式低效的問題,引入了緩衝區的概念,當資料到達時,可以預先被寫入緩衝區,再由緩衝區交給執行緒,因此執行緒無需阻塞地等待IO。
Channel(通道):
用於源節點與目標節點的連線,在java NIO 中負責緩衝區中的資料傳輸,Channel本身不儲存資料,因此需要緩衝區配合進行傳輸,通道就相當於鐵軌,緩衝區就相當於火車。
引用 Java NIO 中權威的說法:通道是 I/O 傳輸發生時通過的入口,而緩衝區是這些數 據傳輸的來源或目標。對於離開緩衝區的傳輸,您想傳遞出去的資料被置於一個緩衝區,被傳送到通道。對於傳回緩衝區的傳輸,一個通道將資料放置在您所提供的緩衝區中。
例如 有一個伺服器通道 ServerSocketChannel serverChannel,一個客戶端通道 SocketChannel clientChannel;伺服器緩衝區:serverBuffer,客戶端緩衝區:clientBuffer。 當伺服器想向客戶端傳送資料時,需要呼叫:clientChannel.write(serverBuffer)。當客戶端要讀時,呼叫 clientChannel.read(clientBuffer) 當客戶端想向伺服器傳送資料時,需要呼叫:serverChannel.write(clientBuffer)。當伺服器要讀時,呼叫 serverChannel.read(serverBuffer)
Selector(選擇器):
通道和緩衝區的機制,使得執行緒無需阻塞地等待IO事件的就緒,但是總是要有人來監管這些IO事件。這個工作就交給了selector來完成,這就是所謂的同步。Selector允許單執行緒處理多個 Channel。如果你的應用打開了多個連線(通道),但每個連線的流量都很低,使用Selector就會很方便。要使用Selector,得向Selector註冊Channel,然後呼叫它的select()方法。這個方法會一直阻塞到某個註冊的通道有事件就緒,這就是所說的輪詢。一旦這個方法返回,執行緒就可以處理這些事件。
Selector中註冊的感興趣事件有:
OP_ACCEPT:接收
OP_CONNECT :連線
OP_READ :讀
OP_WRITE:寫
二、Java NIO 常用API
1.Buffer
package com.jxhtyw;
import java.nio.ByteBuffer;
import org.junit.Test;
/**
*緩衝區(buffer):在java NIO中負責資料的存取,緩衝區就是陣列,用於儲存不同型別的緩衝區
*緩衝區型別:
*ByteBuffer
*CharBuffer
*IntBuffer
*ShortBuffer
*LongBuffer
*DoubleBuffer
*
* 核心方法:put()和get()
*
* 核心屬性:
* capacity:容量,表示緩衝區中最大資料的儲存容量
* limit:界限,表示緩衝區中可以操作資料的大小(limit後資料不能進行讀寫)
* position:位置,表示緩衝區中正在操作資料的位置
*
* mark():標記
* reset():恢復到mark的位置
*
* @author dxx
* @version 2017-11-02 下午7:46:38
*/
public class NioBuffer {
String str="helloworld";
@Test
public void run(){
//1.分配一個指定大小的緩衝區
ByteBuffer bb=ByteBuffer.allocate(1024);
System.out.println(bb.capacity());
System.out.println(bb.limit());
System.out.println(bb.position());
//2.存入資料
bb.put(str.getBytes());
System.out.println(bb.capacity());
System.out.println(bb.limit());
System.out.println(bb.position());//位置變為10
//3.切換讀取資料模式
bb.flip();
System.out.println(bb.capacity());//緩衝區大小仍然為1024
System.out.println(bb.limit());//可讀取數量為10個的位元組
System.out.println(bb.position());//位置切換到0了,可以從0開始讀取
//4.讀取資料
byte[] by=new byte[bb.limit()];
bb.get(by);//獲取到緩衝區可讀取的所有資料(也就是10),存放在by陣列中
//System.out.println(by);
System.out.println(new String(by,0,by.length));
System.out.println(bb.capacity());
System.out.println(bb.limit());
System.out.println(bb.position());
try {
bb.get();//再讀的話就越界了
} catch (Exception e) {
e.printStackTrace();
}
//5.rewind() :可重複讀資料
bb.rewind();
System.out.println(bb.capacity());
System.out.println(bb.limit());
System.out.println(bb.position());//位置變為0了,說明又可以讀了
//6.clear():清空緩衝區,但是緩衝區的資料依然存在,但是處於“被遺忘狀態”
bb.clear();
System.out.println(bb.capacity());
System.out.println(bb.limit());//指標全部回到最原始狀態,不知道有多少資料
System.out.println(bb.position());
System.out.println((char)bb.get());
}
@Test
public void run2(){
ByteBuffer bb=ByteBuffer.allocate(1024);
bb.put(str.getBytes());
bb.flip();
byte[] by=new byte[bb.limit()];
bb.get(by,0,2);
System.out.println(new String(by,0,2));
System.out.println(bb.position());//到第二個位元組了
//標記
bb.mark();
bb.get(by,2,3);
System.out.println(new String(by,2,3));
System.out.println(bb.position());//到第二個位元組了
//重置
bb.reset();
System.out.println(bb.position());//位置又回到標記處
}
@Test
public void run3(){
//非直接緩衝區:通過allocate()方法分配緩衝區,將緩衝區建立在JVM的記憶體中
ByteBuffer bb=ByteBuffer.allocate(1024);
//直接緩衝區:通過allocateDirect()方法分配直接緩衝區,將緩衝區建立在實體記憶體中,可以提高效率
ByteBuffer bb2=ByteBuffer.allocateDirect(1024);
}
}
2.Channel
package com.jxhtyw;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.FileChannel.MapMode;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import org.junit.Test;
/**
*通道(Channel):用於源節點與目標節點的連線,在java NIO 中負責緩衝區中的資料傳輸,
*Channel本身不儲存資料,因此需要緩衝區配合進行傳輸,通道就相當於鐵軌,緩衝區就相當於火車
*
*JDK 1.7中的NIO 針對各個通道提供了靜態方法open()
*
*通道之間的資料傳輸
*transferTo()
*transferFrom()
*
*
*分散與聚集
*分散讀取:將通道中的資料分散到多個緩衝區中去
*聚集寫入:將多個緩衝區的資料聚集到通道中去
*
* @author dxx
* @version 2017-11-02 下午8:33:05
*/
public class NioChannel {
//4.分散與聚集
@Test
public void run4() throws IOException{
//1.分散讀取
RandomAccessFile raf1=new RandomAccessFile("1.txt", "rw");
//獲取通道
FileChannel channel1 = raf1.getChannel();
//分配兩個指定大小的緩衝區
ByteBuffer buf1 = ByteBuffer.allocate(100);
ByteBuffer buf2 = ByteBuffer.allocate(1024);
//構建緩衝區陣列
ByteBuffer[] bufArr={buf1,buf2};
//通道讀取
channel1.read(bufArr);
//切換緩衝區為寫模式
for (ByteBuffer byteBuffer : bufArr) {
byteBuffer.flip();
}
System.out.println(new String(bufArr[0].array(), 0, bufArr[0].limit()));
System.out.println("--------------------------");
System.out.println(new String(bufArr[1].array(), 0, bufArr[1].limit()));
//2.聚集寫入
//聚集寫入到2.txt中
RandomAccessFile raf2=new RandomAccessFile("2.txt", "rw");
FileChannel channel2 = raf2.getChannel();
//將緩衝區陣列寫入通道中
channel2.write(bufArr);
}
//3.通道之間的資料傳輸,直接緩衝區
@Test
public void run3() throws IOException{
FileChannel inChannel = FileChannel.open(Paths.get("e:/01.avi"), StandardOpenOption.READ);
FileChannel outChannel = FileChannel.open(Paths.get("e:/02.avi"), StandardOpenOption.WRITE,StandardOpenOption.READ,StandardOpenOption.CREATE);
// inChannel.transferTo(0, inChannel.size(), outChannel);
outChannel.transferFrom(inChannel, 0, inChannel.size());
}
//2.利用直接緩衝區完成檔案的複製
@Test
public void run2() throws IOException{
long start=System.currentTimeMillis();
//獲取通道
//讀模式
FileChannel inChannel = FileChannel.open(Paths.get("e:/01.avi"), StandardOpenOption.READ);
//讀寫模式
FileChannel outChannel = FileChannel.open(Paths.get("e:/02.avi"), StandardOpenOption.WRITE,StandardOpenOption.READ,StandardOpenOption.CREATE);
//記憶體對映檔案
MappedByteBuffer inBuf = inChannel.map(MapMode.READ_ONLY, 0, inChannel.size());
MappedByteBuffer outBuf = outChannel.map(MapMode.READ_WRITE,0,inChannel.size());
//對緩衝區的資料進行讀寫操作
byte[] by=new byte[inBuf.limit()];
inBuf.get(by);
outBuf.put(by);
inChannel.close();
outChannel.close();
long end=System.currentTimeMillis();
System.out.println("耗費時間:"+(end-start));
}
//1.利用通道完成檔案的複製(非直接緩衝區)
@Test
public void run() throws IOException{
long start=System.currentTimeMillis();
FileInputStream fis=null;
FileOutputStream fos=null;
FileChannel inChannel = null;
FileChannel outChannel = null;
try {
fis=new FileInputStream("e:/01.avi");
fos=new FileOutputStream("e:/02.avi");
inChannel = fis.getChannel();
outChannel = fos.getChannel();
//分配指定大小的緩衝區
ByteBuffer bb = ByteBuffer.allocate(1024);
//將通道中的資料存入緩衝區,這個時候的緩衝區是寫模式
while(inChannel.read(bb)!=-1){
//將緩衝區切換為讀模式
bb.flip();
outChannel.write(bb);
bb.clear();
}
} catch (Exception e) {
e.printStackTrace();
}finally{
fis.close();
fos.close();
inChannel.close();
outChannel.close();
}
long end=System.currentTimeMillis();
System.out.println("耗費時間:"+(end-start));
}
}
3.阻塞NIO
package com.jxhtyw;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import org.junit.Test;
/**
* 阻塞NIO
*
* @author dxx
* @version 2017-11-02 下午9:36:08
*/
public class NioBlock2 {
//1.網路通訊客戶端
@Test
public void client() throws IOException{
//獲取通道
SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress("127.0.0.1", 8888));
FileChannel fChannel = FileChannel.open(Paths.get("1.png"), StandardOpenOption.READ);
//分配指定大小的緩衝區
ByteBuffer bb = ByteBuffer.allocate(1024);
//讀取本地檔案,併發送到客戶端
while(fChannel.read(bb)!=-1){
//將緩衝區轉換為寫模式
bb.flip();
socketChannel.write(bb);
bb.clear();
}
//當接收不到緩衝區的資料時,強制關閉通道(阻塞式體現在這裡)
socketChannel.shutdownOutput();
//接受伺服器端的反饋
int len=0;
while((len=socketChannel.read(bb))!=-1){
bb.flip();
System.out.println(new String(bb.array(),0,len));
bb.clear();
}
//關閉通道
socketChannel.close();
fChannel.close();
}
//2.服務端
@Test
public void server() throws IOException{
//獲取通道
ServerSocketChannel ssChannel = ServerSocketChannel.open();
FileChannel fChannel = FileChannel.open(Paths.get("2.png"), StandardOpenOption.WRITE,StandardOpenOption.CREATE);
//繫結連線
ssChannel.bind(new InetSocketAddress(8888));
//獲取客戶端連線的通道
SocketChannel accept = ssChannel.accept();
//分配指定大小的緩衝區
ByteBuffer bb = ByteBuffer.allocate(1024);
//讀取本地檔案,併發送到客戶端
while(accept.read(bb)!=-1){
//將緩衝區轉換為寫模式
bb.flip();
fChannel.write(bb);
bb.clear();
}
//傳送反饋給客戶端
bb.put("接收資料成功".getBytes());
bb.flip();
accept.write(bb);
//關閉通道
ssChannel.close();
fChannel.close();
}
}
執行結果:
3.非阻塞NIO——簡單聊天室的建立
package com.jxhtyw;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Iterator;
import java.util.Random;
import java.util.Scanner;
import org.junit.Test;
/**
* 非阻塞NIO
*
*一、使用NIO完成網路通訊
*1.通道(Channel):負責連線
*2.緩衝區(Buffer):負責資料的存取
*3.選擇器(Selector):用於監控IO情況
*
* @author dxx
* @version 2017-11-05 上午10:42:18
*/
public class NioNotBlock {
//客戶端
@Test
public void client() throws IOException{
//1.獲取通道
SocketChannel sChannel = SocketChannel.open(new InetSocketAddress("127.0.0.1", 8888));
//2.切換為非阻塞模式
sChannel.configureBlocking(false);
//3.分配指定大小的緩衝區
ByteBuffer bb = ByteBuffer.allocate(1024);
//4.傳送資料到客戶端(實時時間+鍵盤錄入)
Scanner sc=new Scanner(System.in);
String[] nameArr={"Kobe","James","Iverson","Duncan","Curry","Harden"};
Random rand = new Random();
int i = rand.nextInt(5);
SimpleDateFormat sdf=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
String date = sdf.format(new Date());
while(sc.hasNext()){
String str = sc.next();
bb.put((nameArr[i]+" "+date+"\n"+str).getBytes());
bb.flip();
sChannel.write(bb);
bb.clear();
}
//5.關閉通道
sChannel.close();
}
//服務端
@Test
public void server() throws IOException{
//1.獲取通道
ServerSocketChannel ssChannel = ServerSocketChannel.open();
//2.切換為非阻塞模式
ssChannel.configureBlocking(false);
//3.繫結連線
ssChannel.bind(new InetSocketAddress(8888));
//4.獲取選擇器
Selector selector=Selector.open();
//5.將通道註冊到選擇器上,指定“監聽接收事件”
ssChannel.register(selector, SelectionKey.OP_ACCEPT);
//6.輪詢獲取選擇器上已經“準備就緒”的事件
while(selector.select()>0){
//7.獲取當前選擇器中所有註冊的“選擇鍵(已就緒的監聽事件)”
Iterator<SelectionKey> it = selector.selectedKeys().iterator();
while(it.hasNext()){
//8.獲取準備就緒的事件
SelectionKey sk = it.next();
//9.判斷是什麼事件準備就緒
if(sk.isAcceptable()){
//10.若是接收就緒,則獲取客戶端連線
SocketChannel clientCh = ssChannel.accept();
//11.切換非阻塞模式
clientCh.configureBlocking(false);
//12.將該通道註冊到選擇器
clientCh.register(selector,SelectionKey.OP_READ);
}else if(sk.isReadable()){
//13.獲取當前選擇器上“讀就緒”的通道
SocketChannel rChannel=(SocketChannel) sk.channel();
//14.讀取資料
ByteBuffer bb = ByteBuffer.allocate(1024);
int len=0;
while((len=rChannel.read(bb))>0){
bb.flip();
System.out.println(new String(bb.array(),0,len));
bb.clear();
}
}
//15.取消選擇鍵 SelectionKey
it.remove();
}
}
}
}
聊天室效果: