posts - 56,  comments - 12,  trackbacks - 0

          JDK 1.4開始,Java的標準庫中就包含了NIO, 即所謂的“New IO”。其中最重要的功能就是提供了“非阻塞”的IO,當然包括了Socket。NonBlocking的IO就是對select(Unix平臺下)以及 WaitForMultipleObjects(Windows平臺)的封裝,提供了高性能、易伸縮的服務架構。

          說來慚愧,直到JDK1.4才有這種功能,但遲到者不一定沒有螃蟹吃,NIO就提供了優秀的面向對象的解決方案,可以很方便地編寫高性能的服務器。

          話說回來,傳統的Server/Client實現是基于Thread per request,即服務器為每個客戶端請求建立一個線程處理,單獨負責處理一個客戶的請求。比如像Tomcat(新版本也會提供NIO方案)、Resin等Web服務器就是這樣實現的。當然為了減少瞬間峰值問題,服務器一般都使用線程池,規定了同時并發的最大數量,避免了線程的無限增長。

          但這樣有一個問題:如果線程池的大小為100,當有100個用戶同時通過HTTP現在一個大文件時,服務器的線程池會用完,因為所有的線程都在傳輸大文件了,即使第101個請求者僅僅請求一個只有10字節的頁面,服務器也無法響應了,只有等到線程池中有空閑的線程出現。

          另外,線程的開銷也是很大的,特別是達到了一個臨界值后,性能會顯著下降,這也限制了傳統的Socket方案無法應對并發量大的場合,而“非阻塞”的IO就能輕松解決這個問題。

          下面只是一個簡單的例子:服務器提供了下載大型文件的功能,客戶端連接上服務器的12345端口后,就可以讀取服務器發送的文件內容信息了。注意這里的服務器只有一個主線程,沒有其他任何派生線程,讓我們看看NIO是如何用一個線程處理N個請求的。

          NIO服務器最核心的一點就是反應器模式:當有感興趣的事件發生的,就通知對 應的事件處理器去處理這個事件,如果沒有,則不處理。所以使用一個線程做輪詢就可以了。當然這里這是個例子,如果要獲得更高性能,可以使用少量的線程,一 個負責接收請求,其他的負責處理請求,特別是對于多CPU時效率會更高。

          關于使用NIO過程中出現的問題,最為普遍的就是為什么沒有請求時CPU的占 用率為100%?出現這種問題的主要原因是注冊了不感興趣的事件,比如如果沒有數據要發到客戶端,而又注冊了寫事件(OP_WRITE),則在 Selector.select()上就會始終有事件出現,CPU就一直處理了,而此時select()應該是阻塞的。

          另外一個值得注意的問題是:由于只使用了一個線程(多個線程也如此)處理用戶請求,所以要避免線程被阻塞,解決方法是事件的處理者必須要即刻返回,不能陷入循環中,否則會影響其他用戶的請求速度。

          具體到本例子中,由于文件比較大,如果一次性發送整個文件(這里的一次性不是 指send整個文件內容,而是通過while循環不間斷的發送分組包),則主線程就會阻塞,其他用戶就不能響應了。這里的解決方法是當有WRITE事件 時,僅僅是發送一個塊(比如4K字節)。發完后,繼續等待WRITE事件出現,依次處理,直到整個文件發送完畢,這樣就不會阻塞其他用戶了。

          服務器的例子:

          package  nio.file;

          import  java.io.FileInputStream;
          import  java.io.IOException;
          import  java.net.InetSocketAddress;
          import  java.nio.ByteBuffer;
          import  java.nio.CharBuffer;
          import  java.nio.channels.FileChannel;
          import  java.nio.channels.Selecti;
          import  java.nio.channels.Selector;
          import  java.nio.channels.ServerSocketChannel;
          import  java.nio.channels.SocketChannel;
          import  java.nio.charset.Charset;
          import  java.nio.charset.CharsetDecoder;
          import  java.util.Iterator;

          /**
            * 測試文件下載的NIOServer
           
            @author  tenyears.cn
            */
          public class  NIOServer  {
             static  int  BLOCK =  4096 ;
             // 處理與客戶端的交互
             public class  HandleClient  {
               protected  FileChannel channel;
               protected  ByteBuffer buffer;
               public  HandleClient ()  throws  IOException  {
                 this .channel =  new  FileInputStream ( filename ) .getChannel () ;
                 this .buffer = ByteBuffer.allocate ( BLOCK ) ;
               }
               public  ByteBuffer readBlock () {
                 try  {
                   buffer.clear () ;
                   int  count = channel.read ( buffer ) ;
                   buffer.flip () ;
                   if  ( count <=  0 )
                     return null ;
                 catch  ( IOException e ) {
                   e.printStackTrace () ;
                 }
                 return  buffer;
               }
               public  void  close () {
                 try  {
                   channel.close () ;
                 catch  ( IOException e ) {
                   e.printStackTrace () ;
                 }
               }
             }

             protected  Selector selector;
             protected  String filename =  "d:\\bigfile.dat" // a big file
             protected  ByteBuffer clientBuffer = ByteBuffer.allocate ( BLOCK ) ;
             protected  CharsetDecoder decoder;

             public  NIOServer ( int  port throws  IOException  {
               selector =  this .getSelector ( port ) ;
               Charset charset = Charset.forName ( "GB2312" ) ;
               decoder = charset.newDecoder () ;
             }

             // 獲取Selector
             protected  Selector getSelector ( int  port throws  IOException  {
               ServerSocketChannel server = ServerSocketChannel.open () ;
               Selector sel = Selector.open () ;
               server.socket () .bind ( new  InetSocketAddress ( port )) ;
               server.configureBlocking ( false ) ;
               server.register ( sel, Selecti.OP_ACCEPT ) ;
               return  sel;
             }

             // 監聽端口
             public  void  listen () {
               try  {
                 for  ( ;; ) {
                   selector.select () ;
                   Iterator iter = selector.selectedKeys ()
                       .iterator () ;
                   while  ( iter.hasNext ()) {
                     Selecti key = iter.next () ;
                     iter.remove () ;
                     handleKey ( key ) ;
                   }
                 }
               catch  ( IOException e ) {
                 e.printStackTrace () ;
               }
             }

             // 處理事件
             protected  void  handleKey ( Selecti key throws  IOException  {
               if  ( key.isAcceptable ()) {  // 接收請求
                 ServerSocketChannel server =  ( ServerSocketChannel key.channel () ;
                 SocketChannel channel = server.accept () ;
                 channel.configureBlocking ( false ) ;
                 channel.register ( selector, Selecti.OP_READ ) ;
               else if  ( key.isReadable ()) {  // 讀信息
                 SocketChannel channel =  ( SocketChannel key.channel () ;
                 int  count = channel.read ( clientBuffer ) ;
                 if  ( count >  0 ) {
                   clientBuffer.flip () ;
                   CharBuffer charBuffer = decoder.decode ( clientBuffer ) ;
                   System.out.println ( "Client >>"  + charBuffer.toString ()) ;
                   Selecti wKey = channel.register ( selector,
                       Selecti.OP_WRITE ) ;
                   wKey.attach ( new  HandleClient ()) ;
                 else
                   channel.close () ;
                 clientBuffer.clear () ;
               else if  ( key.isWritable ()) {  // 寫事件
                 SocketChannel channel =  ( SocketChannel key.channel () ;
                 HandleClient handle =  ( HandleClient key.attachment () ;
                 ByteBuffer block = handle.readBlock () ;
                 if  ( block !=  null )
                   channel.write ( block ) ;
                 else  {
                   handle.close () ;
                   channel.close () ;
                 }
               }
             }

             public static  void  main ( String []  args ) {
               int  port =  12345 ;
               try  {
                 NIOServer server =  new  NIOServer ( port ) ;
                 System.out.println ( "Listernint on "  + port ) ;
                 while  ( true ) {
                   server.listen () ;
                 }
               catch  ( IOException e ) {
                 e.printStackTrace () ;
               }
             }
          }

          該代碼中,通過一個HandleClient來獲取文件的一塊數據,每一個客戶都會分配一個HandleClient的實例。

          下面是客戶端請求的代碼,也比較簡單,模擬100個用戶同時下載文件。

          package  nio.file;
          import  java.io.IOException;
          import  java.net.InetSocketAddress;
          import  java.nio.ByteBuffer;
          import  java.nio.CharBuffer;
          import  java.nio.channels.Selecti;
          import  java.nio.channels.Selector;
          import  java.nio.channels.SocketChannel;
          import  java.nio.charset.Charset;
          import  java.nio.charset.CharsetEncoder;
          import  java.util.Iterator;
          import  java.util.concurrent.ExecutorService;
          import  java.util.concurrent.Executors;

          /**
            * 文件下載客戶端
            @author  tenyears.cn
            */
          public class  NIOClient  {
             static  int  SIZE =  100 ;
             static  InetSocketAddress ip =  new  InetSocketAddress ( "localhost" , 12345 ) ;
             static  CharsetEncoder encoder = Charset.forName ( "GB2312" ) .newEncoder () ;
             static class  Download  implements  Runnable  {
               protected  int  index;
               public  Download ( int  index ) {
                 this .index = index;
               }

               public  void  run () {
                 try  {
                   long  start = System.currentTimeMillis () ;
                   SocketChannel client = SocketChannel.open () ;
                   client.configureBlocking ( false ) ;
                   Selector selector = Selector.open () ;
                   client.register ( selector, Selecti.OP_CONNECT ) ;
                   client.connect ( ip ) ;
                   ByteBuffer buffer = ByteBuffer.allocate ( 1024 ) ;
                   int  total =  0 ;
                   FOR:  for  ( ;; ) {
                     selector.select () ;
                     Iterator iter = selector.selectedKeys ()
                         .iterator () ;
                     while  ( iter.hasNext ()) {
                       Selecti key = iter.next () ;
                       iter.remove () ;
                       if  ( key.isConnectable ()) {
                         SocketChannel channel =  ( SocketChannel key
                             .channel () ;
                         if  ( channel.isConnectionPending ())
                           channel.finishConnect () ;
                         channel.write ( encoder.encode ( CharBuffer
                             .wrap ( "Hello from "  + index ))) ;
                         channel.register ( selector, Selecti.OP_READ ) ;
                       else if  ( key.isReadable ()) {
                         SocketChannel channel =  ( SocketChannel key
                             .channel () ;
                         int  count = channel.read ( buffer ) ;
                         if  ( count >  0 ) {
                           total += count;
                           buffer.clear () ;
                         else  {
                           client.close () ;
                           break  FOR;
                         }
                       }
                     }
                   }
                   double  last =  ( System.currentTimeMillis ()  - start 1.0  1000 ;
                   System.out.println ( "Thread "  + index +  " downloaded "  + total
                       "bytes in "  + last +  "s." ) ;
                 catch  ( IOException e ) {
                   e.printStackTrace () ;
                 }
               }
             }

             public static  void  main ( String []  args throws  IOException  {
               ExecutorService exec = Executors.newFixedThreadPool ( SIZE ) ;
               for  ( int  index =  0 ; index < SIZE; index++ ) {
                 exec.execute ( new  Download ( index )) ;
               }
               exec.shutdown () ;
             }
          }
          posted on 2007-01-19 00:03 苦笑枯 閱讀(471) 評論(0)  編輯  收藏 所屬分類: Java
          收藏來自互聯網,僅供學習。若有侵權,請與我聯系!

          <2007年1月>
          31123456
          78910111213
          14151617181920
          21222324252627
          28293031123
          45678910

          常用鏈接

          留言簿(2)

          隨筆分類(56)

          隨筆檔案(56)

          搜索

          •  

          最新評論

          閱讀排行榜

          評論排行榜

          主站蜘蛛池模板: 眉山市| 辉县市| 东山县| 凤山县| 佛坪县| 申扎县| 盐津县| 邹平县| 马边| 手机| 菏泽市| 偃师市| 诸城市| 如东县| 增城市| 东莞市| 水富县| 广丰县| 建瓯市| 格尔木市| 陈巴尔虎旗| 阿鲁科尔沁旗| 营口市| 花莲县| 游戏| 武鸣县| 安泽县| 同心县| 阜阳市| 灵台县| 宜城市| 楚雄市| 桃江县| 博野县| 沂源县| 桓台县| 梅州市| 台江县| 丹凤县| 丰宁| 新竹市|