通道（Channel）是java.nio的第二個主要創(chuàng)新。它們既不是一個擴展也不是一項增強，而是全新、極好的Java I/O示例，提供與I/O服務(wù)的直接連接。Channel用于在字節(jié)緩沖區(qū)和位于通道另一側(cè)的實體（通常是一個文件或套接字）之間有效地傳輸數(shù)據(jù)。

channel介紹

通道是訪問I/O服務(wù)的導(dǎo)管。I/O可以分為廣義的兩大類別：File I/O和Stream I/O。那么相應(yīng)地有兩種類型的通道也就不足為怪了，它們是文件（file）通道和套接字（socket）通道。我們看到在api里有一個FileChannel類和三個socket通道類：SocketChannel、ServerSocketChannel和DatagramChannel。

通道可以以多種方式創(chuàng)建。Socket通道有可以直接創(chuàng)建新socket通道的工廠方法。但是一個FileChannel對象卻只能通過在一個打開的RandomAccessFile、FileInputStream或FileOutputStream對象上調(diào)用getChannel( )方法來獲取。你不能直接創(chuàng)建一個FileChannel對象。

我們先來看一下FileChannel的用法：

 // 創(chuàng)建文件輸出字節(jié)流
 FileOutputStream fos = new FileOutputStream("data.txt");
 //得到文件通道
 FileChannel fc = fos.getChannel();
 //往通道寫入ByteBuffer
 fc.write(ByteBuffer.wrap("Some text ".getBytes()));
 //關(guān)閉流
 fos.close();

 //隨機訪問文件
 RandomAccessFile raf = new RandomAccessFile("data.txt", "rw");
 //得到文件通道
 fc = raf.getChannel();
 //設(shè)置通道的文件位置 為末尾
 fc.position(fc.size()); 
 //往通道寫入ByteBuffer
 fc.write(ByteBuffer.wrap("Some more".getBytes()));
 //關(guān)閉
 raf.close();

 //創(chuàng)建文件輸入流
 FileInputStream fs = new FileInputStream("data.txt");
 //得到文件通道
 fc = fs.getChannel();
 //分配ByteBuffer空間大小
 ByteBuffer buff = ByteBuffer.allocate(BSIZE);
 //從通道中讀取ByteBuffer
 fc.read(buff);
 //調(diào)用此方法為一系列通道寫入或相對獲取 操作做好準備
 buff.flip();
 //從ByteBuffer從依次讀取字節(jié)并打印
 while (buff.hasRemaining()){
  System.out.print((char) buff.get());
 }
 fs.close();

再來看一下SocketChannel：

 SocketChannel sc = SocketChannel.open( );
 sc.connect (new InetSocketAddress ("somehost", someport)); 
 ServerSocketChannel ssc = ServerSocketChannel.open( ); 
 ssc.socket( ).bind (new InetSocketAddress (somelocalport)); 
 DatagramChannel dc = DatagramChannel.open( );

可以設(shè)置 SocketChannel 為非阻塞模式（non-blocking mode）.設(shè)置之后，就可以在異步模式下調(diào)用connect(), read() 和write()了。如果SocketChannel在非阻塞模式下，此時調(diào)用connect()，該方法可能在連接建立之前就返回了。為了確定連接是否建立，可以調(diào)用finishConnect()的方法。像這樣：

socketChannel.configureBlocking(false);
socketChannel.connect(new InetSocketAddress("http://jenkov.com", 80));

while(! socketChannel.finishConnect() ){
 //wait, or do something else...
}

服務(wù)器端的使用經(jīng)常會考慮到非阻塞socket通道，因為它們使同時管理很多socket通道變得更容易。但是，在客戶端使用一個或幾個非阻塞模式的socket通道也是有益處的，例如，借助非阻塞socket通道，GUI程序可以專注于用戶請求并且同時維護與一個或多個服務(wù)器的會話。在很多程序上，非阻塞模式都是有用的。

調(diào)用finishConnect( )方法來完成連接過程，該方法任何時候都可以安全地進行調(diào)用。假如在一個非阻塞模式的SocketChannel對象上調(diào)用finishConnect( )方法，將可能出現(xiàn)下列情形之一：

• connect( )方法尚未被調(diào)用。那么將產(chǎn)生NoConnectionPendingException異常。
• 連接建立過程正在進行，尚未完成。那么什么都不會發(fā)生，finishConnect( )方法會立即返回false值。
• 在非阻塞模式下調(diào)用connect( )方法之后，SocketChannel又被切換回了阻塞模式。那么如果有必要的話，調(diào)用線程會阻塞直到連接建立完成，finishConnect( )方法接著就會返回true值。在初次調(diào)用connect( )或最后一次調(diào)用finishConnect( )之后，連接建立過程已經(jīng)完成。那么SocketChannel對象的內(nèi)部狀態(tài)將被更新到已連接狀態(tài)，finishConnect( )方法會返回true值，然后SocketChannel對象就可以被用來傳輸數(shù)據(jù)了。
• 連接已經(jīng)建立。那么什么都不會發(fā)生，finishConnect( )方法會返回true值。

Socket通道是線程安全的。并發(fā)訪問時無需特別措施來保護發(fā)起訪問的多個線程，不過任何時候都只有一個讀操作和一個寫操作在進行中。請記住，sockets是面向流的而非包導(dǎo)向的。它們可以保證發(fā)送的字節(jié)會按照順序到達但無法承諾維持字節(jié)分組。某個發(fā)送器可能給一個socket寫入了20個字節(jié)而接收器調(diào)用read( )方法時卻只收到了其中的3個字節(jié)。剩下的17個字節(jié)還是傳輸中。由于這個原因，讓多個不配合的線程共享某個流socket的同一側(cè)絕非一個好的設(shè)計選擇。

最后再看一下DatagramChannel：

最后一個socket通道是DatagramChannel。正如SocketChannel對應(yīng)Socket，ServerSocketChannel對應(yīng)ServerSocket，每一個DatagramChannel對象也有一個關(guān)聯(lián)的DatagramSocket對象。不過原命名模式在此并未適用：“DatagramSocketChannel”顯得有點笨拙，因此采用了簡潔的“DatagramChannel”名稱。

正如SocketChannel模擬連接導(dǎo)向的流協(xié)議（如TCP/IP），DatagramChannel則模擬包導(dǎo)向的無連接協(xié)議（如UDP/IP）：

創(chuàng)建DatagramChannel的模式和創(chuàng)建其他socket通道是一樣的：調(diào)用靜態(tài)的open( )方法來創(chuàng)建一個新實例。新DatagramChannel會有一個可以通過調(diào)用socket( )方法獲取的對等DatagramSocket對象。DatagramChannel對象既可以充當服務(wù)器（監(jiān)聽者）也可以充當客戶端（發(fā)送者）。如果你希望新創(chuàng)建的通道負責(zé)監(jiān)聽，那么通道必須首先被綁定到一個端口或地址/端口組合上。綁定DatagramChannel同綁定一個常規(guī)的DatagramSocket沒什么區(qū)別，都是委托對等socket對象上的API實現(xiàn)的：

 DatagramChannel channel = DatagramChannel.open( );
 DatagramSocket socket = channel.socket( ); 
 socket.bind (new InetSocketAddress (portNumber));

DatagramChannel是無連接的。每個數(shù)據(jù)報（datagram）都是一個自包含的實體，擁有它自己的目的地址及不依賴其他數(shù)據(jù)報的數(shù)據(jù)凈荷。與面向流的的socket不同，DatagramChannel可以發(fā)送單獨的數(shù)據(jù)報給不同的目的地址。同樣，DatagramChannel對象也可以接收來自任意地址的數(shù)據(jù)包。每個到達的數(shù)據(jù)報都含有關(guān)于它來自何處的信息（源地址）。

一個未綁定的DatagramChannel仍能接收數(shù)據(jù)包。當一個底層socket被創(chuàng)建時，一個動態(tài)生成的端口號就會分配給它。綁定行為要求通道關(guān)聯(lián)的端口被設(shè)置為一個特定的值（此過程可能涉及安全檢查或其他驗證）。不論通道是否綁定，所有發(fā)送的包都含有DatagramChannel的源地址（帶端口號）。未綁定的DatagramChannel可以接收發(fā)送給它的端口的包，通常是來回應(yīng)該通道之前發(fā)出的一個包。已綁定的通道接收發(fā)送給它們所綁定的熟知端口（wellknown port）的包。數(shù)據(jù)的實際發(fā)送或接收是通過send( )和receive( )方法來實現(xiàn)的。

注意：假如您提供的ByteBuffer沒有足夠的剩余空間來存放您正在接收的數(shù)據(jù)包，沒有被填充的字節(jié)都會被悄悄地丟棄。

Scatter/Gather

通道提供了一種被稱為Scatter/Gather的重要新功能（有時也被稱為矢量I/O）。它是指在多個緩沖區(qū)上實現(xiàn)一個簡單的I/O操作。對于一個write操作而言，數(shù)據(jù)是從幾個緩沖區(qū)按順序抽取（稱為gather）并沿著通道發(fā)送的。緩沖區(qū)本身并不需要具備這種gather的能力（通常它們也沒有此能力）。該gather過程的效果就好比全部緩沖區(qū)的內(nèi)容被連結(jié)起來，并在發(fā)送數(shù)據(jù)前存放到一個大的緩沖區(qū)中。對于read操作而言，從通道讀取的數(shù)據(jù)會按順序被散布（稱為scatter）到多個緩沖區(qū)，將每個緩沖區(qū)填滿直至通道中的數(shù)據(jù)或者緩沖區(qū)的最大空間被消耗完。

scatter / gather經(jīng)常用于需要將傳輸?shù)臄?shù)據(jù)分開處理的場合，例如傳輸一個由消息頭和消息體組成的消息，你可能會將消息體和消息頭分散到不同的buffer中，這樣你可以方便的處理消息頭和消息體。

Scattering Reads是指數(shù)據(jù)從一個channel讀取到多個buffer中。如下圖描述：

代碼示例如下：

ByteBuffer header = ByteBuffer.allocateDirect (10); 
ByteBuffer body = ByteBuffer.allocateDirect (80); 
ByteBuffer [] buffers = { header, body }; 
int bytesRead = channel.read (buffers);

Gathering Writes是指數(shù)據(jù)從多個buffer寫入到同一個channel。如下圖描述：

代碼示例如下：

 ByteBuffer header = ByteBuffer.allocateDirect (10); 
 ByteBuffer body = ByteBuffer.allocateDirect (80); 
 ByteBuffer [] buffers = { header, body }; 
 channel.write(bufferArray);

使用得當?shù)脑?，Scatter/Gather會是一個極其強大的工具。它允許你委托操作系統(tǒng)來完成辛苦活：將讀取到的數(shù)據(jù)分開存放到多個存儲桶（bucket）或者將不同的數(shù)據(jù)區(qū)塊合并成一個整體。這是一個巨大的成就，因為操作系統(tǒng)已經(jīng)被高度優(yōu)化來完成此類工作了。它節(jié)省了您來回移動數(shù)據(jù)的工作，也就避免了緩沖區(qū)拷貝和減少了您需要編寫、調(diào)試的代碼數(shù)量。既然您基本上通過提供數(shù)據(jù)容器引用來組合數(shù)據(jù)，那么按照不同的組合構(gòu)建多個緩沖區(qū)陣列引用，各種數(shù)據(jù)區(qū)塊就可以以不同的方式來組合了。下面的例子好地詮釋了這一點：

public class GatheringTest {
 private static final String DEMOGRAPHIC = "output.txt";
 public static void main (String [] argv) throws Exception {
  int reps = 10;
  if (argv.length > 0) {
   reps = Integer.parseInt(argv[0]);
  }
  FileOutputStream fos = new FileOutputStream(DEMOGRAPHIC);
  GatheringByteChannel gatherChannel = fos.getChannel();

  ByteBuffer[] bs = utterBS(reps);

  while (gatherChannel.write(bs) > 0) {
   // 不做操作，讓通道把數(shù)據(jù)輸出到文件寫完
  }
  System.out.println("Mindshare paradigms synergized to " + DEMOGRAPHIC);
  fos.close();
 }
 private static String [] col1 = { "Aggregate", "Enable", "Leverage",
          "Facilitate", "Synergize", "Repurpose",
          "Strategize", "Reinvent", "Harness"
         };

 private static String [] col2 = { "cross-platform", "best-of-breed", "frictionless",
          "ubiquitous", "extensible", "compelling",
          "mission-critical", "collaborative", "integrated"
         };

 private static String [] col3 = { "methodologies", "infomediaries", "platforms", "schemas", "mindshare", "paradigms", "functionalities", "web services", "infrastructures" };

 private static String newline = System.getProperty ("line.separator");


 private static ByteBuffer [] utterBS (int howMany) throws Exception {
  List list = new LinkedList();
  for (int i = 0; i < howMany; i++) {
   list.add(pickRandom(col1, " "));
   list.add(pickRandom(col2, " "));
   list.add(pickRandom(col3, newline));
  }
  ByteBuffer[] bufs = new ByteBuffer[list.size()];
  list.toArray(bufs);
  return (bufs);
 }
 private static Random rand = new Random( );


 /**
  * 隨機生成字符
  * @param strings
  * @param suffix
  * @return
  * @throws Exception
  */
 private static ByteBuffer pickRandom (String [] strings, String suffix) throws Exception {
  String string = strings [rand.nextInt (strings.length)];
  int total = string.length() + suffix.length( );
  ByteBuffer buf = ByteBuffer.allocate (total);
  buf.put (string.getBytes ("US-ASCII"));
  buf.put (suffix.getBytes ("US-ASCII"));
  buf.flip( );
  return (buf);
 }
}

輸出為：

 Reinvent integrated web services
 Aggregate best-of-breed platforms
 Harness frictionless platforms
 Repurpose extensible paradigms
 Facilitate ubiquitous methodologies
 Repurpose integrated methodologies
 Facilitate mission-critical paradigms
 Synergize compelling methodologies
 Reinvent compelling functionalities
 Facilitate extensible platforms

雖然這種輸出沒有什么意義，但是gather確是很容易的讓我們把它輸出出來。

Pipe

java.nio.channels包中含有一個名為Pipe（管道）的類。廣義上講，管道就是一個用來在兩個實體之間單向傳輸數(shù)據(jù)的導(dǎo)管。
Java NIO 管道是2個線程之間的單向數(shù)據(jù)連接。Pipe有一個source通道和一個sink通道。數(shù)據(jù)會被寫到sink通道，從source通道讀取。Pipe類創(chuàng)建一對提供環(huán)回機制的Channel對象。這兩個通道的遠端是連接起來的，以便任何寫在SinkChannel對象上的數(shù)據(jù)都能出現(xiàn)在SourceChannel對象上。

下面我們來創(chuàng)建一條Pipe，并向Pipe中寫數(shù)據(jù)：

//通過Pipe.open()方法打開管道
Pipe pipe = Pipe.open();

//要向管道寫數(shù)據(jù)，需要訪問sink通道
Pipe.SinkChannel sinkChannel = pipe.sink();

//通過調(diào)用SinkChannel的write()方法，將數(shù)據(jù)寫入SinkChannel
String newData = "New String to write to file..." + System.currentTimeMillis();
ByteBuffer buf = ByteBuffer.allocate(48);
buf.clear();
buf.put(newData.getBytes());
buf.flip();
while(buf.hasRemaining()) {
 sinkChannel.write(buf);
}

再看如何從管道中讀取數(shù)據(jù)：

讀取管道的數(shù)據(jù)，需要訪問source通道：

Pipe.SourceChannel sourceChannel = pipe.source();

調(diào)用source通道的read()方法來讀取數(shù)據(jù)：

ByteBuffer buf = ByteBuffer.allocate(48);
int bytesRead = sourceChannel.read(buf);

read()方法返回的int值會告訴我們多少字節(jié)被讀進了緩沖區(qū)。

到此我們就把通道的簡單用法講完了，要想會用還是得多去練習(xí)，多模擬使用，這樣才知道什么時候用以及怎么用，下節(jié)我們來講選擇器-Selectors。

以上就是詳解java NIO之Channel（通道）的詳細內(nèi)容，更多關(guān)于JAVA NIO channel（通道）的資料請關(guān)注腳本之家其它相關(guān)文章！

最新評論