posts - 56,  comments - 12,  trackbacks - 0

          本文可看成是對(duì)Doug Lea Scalable IO in Java一文的翻譯。

          當(dāng)前分布式計(jì)算 Web Services盛行天下,這些網(wǎng)絡(luò)服務(wù)的底層都離不開(kāi)對(duì)socket的操作。他們都有一個(gè)共同的結(jié)構(gòu):
          1. Read request
          2. Decode request
          3. Process service
          4. Encode reply
          5. Send reply

          經(jīng)典的網(wǎng)絡(luò)服務(wù)的設(shè)計(jì)如下圖,在每個(gè)線(xiàn)程中完成對(duì)數(shù)據(jù)的處理:

          但這種模式在用戶(hù)負(fù)載增加時(shí),性能將下降非常的快。我們需要重新尋找一個(gè)新的方案,保持?jǐn)?shù)據(jù)處理的流暢,很顯然,事件觸發(fā)機(jī)制是最好的解決辦法,當(dāng)有事件發(fā)生時(shí),會(huì)觸動(dòng)handler,然后開(kāi)始數(shù)據(jù)的處理。

          Reactor模式類(lèi)似于AWT中的Event處理:

          Reactor模式參與者

          1.Reactor 負(fù)責(zé)響應(yīng)IO事件,一旦發(fā)生,廣播發(fā)送給相應(yīng)的Handler去處理,這類(lèi)似于AWT的thread
          2.Handler 是負(fù)責(zé)非堵塞行為,類(lèi)似于AWT ActionListeners;同時(shí)負(fù)責(zé)將handlers與event事件綁定,類(lèi)似于AWT addActionListener

          如圖:

          Java的NIO為reactor模式提供了實(shí)現(xiàn)的基礎(chǔ)機(jī)制,它的Selector當(dāng)發(fā)現(xiàn)某個(gè)channel有數(shù)據(jù)時(shí),會(huì)通過(guò)SlectorKey來(lái)告知我們,在此我們實(shí)現(xiàn)事件和handler的綁定。

          我們來(lái)看看Reactor模式代碼:


          public class Reactor implements Runnable{

            final Selector selector;
          final ServerSocketChannel serverSocket;

            Reactor(int port) throws IOException {
          selector = Selector.open();
          serverSocket = ServerSocketChannel.open();
          InetSocketAddress address = new InetSocketAddress(InetAddress.getLocalHost(),port);
          serverSocket.socket().bind(address);

          serverSocket.configureBlocking(false);
          //向selector注冊(cè)該channel
          Selecti sk =serverSocket.register(selector,Selecti.OP_ACCEPT);

          logger.debug("-->Start serverSocket.register!");

          //利用sk的attache功能綁定Acceptor 如果有事情,觸發(fā)Acceptor
          sk.attach(new Acceptor());
          logger.debug("-->attach(new Acceptor()!");
          }


          public void run() { // normally in a new Thread
          try {
          while (!Thread.interrupted())
          {
          selector.select();
          Set selected = selector.selectedKeys();
          Iterator it = selected.iterator();
          //Selector如果發(fā)現(xiàn)channel有OP_ACCEPT或READ事件發(fā)生,下列遍歷就會(huì)進(jìn)行。
          while (it.hasNext())
          //來(lái)一個(gè)事件 第一次觸發(fā)一個(gè)accepter線(xiàn)程
          //以后觸發(fā)SocketReadHandler
          dispatch((Selecti)(it.next()));
          selected.clear();
          }
          }catch (IOException ex) {
          logger.debug("reactor stop!"+ex);
          }
          }

            //運(yùn)行Acceptor或SocketReadHandler
          void dispatch(Selecti k) {
          Runnable r = (Runnable)(k.attachment());
          if (r != null){
          // r.run();

              }
          }

            class Acceptor implements Runnable { // inner
          public void run() {
          try {
          logger.debug("-->ready for accept!");
          SocketChannel c = serverSocket.accept();
          if (c != null)
          //調(diào)用Handler來(lái)處理channel
          new SocketReadHandler(selector, c);
          }
          catch(IOException ex) {
          logger.debug("accept stop!"+ex);
          }
          }
          }
          }

          以上代碼中巧妙使用了SocketChannel的attach功能,將Hanlder和可能會(huì)發(fā)生事件的channel鏈接在一起,當(dāng)發(fā)生事件時(shí),可以立即觸發(fā)相應(yīng)鏈接的Handler。

          再看看Handler代碼:

          public class SocketReadHandler implements Runnable {

            public static Logger logger = Logger.getLogger(SocketReadHandler.class);

            private Test test=new Test();

            final SocketChannel socket;
          final Selecti sk;

          static final int READING = 0, SENDING = 1;
          int state = READING;

            public SocketReadHandler(Selector sel, SocketChannel c)
          throws IOException {

              socket = c;

              socket.configureBlocking(false);
          sk = socket.register(sel, 0);

              //將Selecti綁定為本Handler 下一步有事件觸發(fā)時(shí),將調(diào)用本類(lèi)的run方法。
          //參看dispatch(Selecti k)
          sk.attach(this);

          //同時(shí)將Selecti標(biāo)記為可讀,以便讀取。
          sk.interestOps(Selecti.OP_READ);
          sel.wakeup();
          }

            public void run() {
          try{
          // test.read(socket,input);
          readRequest() ;
          }catch(Exception ex){
          logger.debug("readRequest error"+ex);
          }
          }


          /**
          * 處理讀取data
          * @param key
          * @throws Exception
          */
          private void readRequest() throws Exception {

            ByteBuffer input = ByteBuffer.allocate(1024);
          input.clear();
          try{

              int bytesRead = socket.read(input);

              ......

              //激活線(xiàn)程池 處理這些request
          requestHandle(new Request(socket,btt));

              .....


          }catch(Exception e) {
          }

          }

          注意在Handler里面又執(zhí)行了一次attach,這樣,覆蓋前面的Acceptor,下次該Handler又有READ事件發(fā)生時(shí),將直接觸發(fā)Handler.從而開(kāi)始了數(shù)據(jù)的讀 處理 寫(xiě) 發(fā)出等流程處理。

          將數(shù)據(jù)讀出后,可以將這些數(shù)據(jù)處理線(xiàn)程做成一個(gè)線(xiàn)程池,這樣,數(shù)據(jù)讀出后,立即扔到線(xiàn)程池中,這樣加速處理速度:

          更進(jìn)一步,我們可以使用多個(gè)Selector分別處理連接和讀事件。

          一個(gè)高性能的Java網(wǎng)絡(luò)服務(wù)機(jī)制就要形成,激動(dòng)人心的集群并行計(jì)算即將實(shí)現(xiàn)。

          posted on 2007-01-18 23:59 苦笑枯 閱讀(328) 評(píng)論(0)  編輯  收藏 所屬分類(lèi): Java
          收藏來(lái)自互聯(lián)網(wǎng),僅供學(xué)習(xí)。若有侵權(quán),請(qǐng)與我聯(lián)系!

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

          常用鏈接

          留言簿(2)

          隨筆分類(lèi)(56)

          隨筆檔案(56)

          搜索

          •  

          最新評(píng)論

          閱讀排行榜

          評(píng)論排行榜

          主站蜘蛛池模板: 宁安市| 公安县| 西吉县| 阿图什市| 泰来县| 江永县| 额济纳旗| 泾源县| 洮南市| 名山县| 珲春市| 平安县| 大洼县| 绍兴市| 邹城市| 岐山县| 马关县| 砚山县| 灵山县| 隆安县| 卫辉市| 宣城市| 沛县| 东丰县| 廊坊市| 和顺县| 玉门市| 平顶山市| 靖安县| 盖州市| 亚东县| 阿城市| 库车县| 东阳市| 鄯善县| 任丘市| 农安县| 吉安市| 博罗县| 太白县| 瑞丽市|