一個高可擴展的基于非阻塞IO的服務器架構

jopen 10年前發布 | 24K 次閱讀架構 Java開發

目錄
線程體系結構
反應堆模式
組件架構
接收器
分配器
分配器級別事件處理器
應用程序級別事件處理器
總結
參考資料

如果你被要求去寫一個高可擴展性的基于JAVA的服務器,你很快就會決定使用JAVA NIO包。為了讓服務器跑起來，你可能會花很多時間閱讀博客和教程來了解線程同步需要NIO SELECTOR類以及處理一些常見的陷阱。本文描述了一個面向連接基于NIO的服務器的基本架構。本文會先看一下一個首選的線程模型然后討論服務器的一些基本組件。

Threading Architecture線程體系結構

第一種也是最直觀的方式去實現一個多線程的服務器是每個連接一個線程的方式。這是JAVA1.4以前的解決方案，由于老版本的JAVA缺少非阻塞的I/O支持。每個連接一個線程的方法分配一個獨家的工作線程給每個連接。在處理循環中，工作線程等待新進入的數據，處理這個請求，返回響應數據，然后再調用阻塞socket的read方法。

public class Server {
  private ExecutorService executors = Executors.newFixedThreadPool(10);
  private boolean isRunning = true;

  public static void main(String... args) throws ... {
    new Server().launch(Integer.parseInt(args[0]));
  } 

  public void launch(int port) throws ... {
    ServerSocket sso = new ServerSocket(port);
    while (isRunning) {
      Socket s = sso.accept();
      executors.execute(new Worker(s));
    }
  }

  private class Worker implements Runnable {
    private LineNumberReader in = null;
    ...

    Worker(Socket s) throws ... {
      in = new LineNumberReader(new InputStreamReader(...));
      out = ...
    }

    public void run() {
      while (isRunning) {
        try {
          // blocking read of a request (line)
          String request = in.readLine();

          // processing the request
          ...
          String response = ...

          // return the response
          out.write(resonse);
          out.flush();
        } catch (Exception e ) {
          ...
        }
      }
      in.close();
      ...
    }
  }
}

在同時發生的客戶端連接和多個同步工作線程之間通常有一個單對單的關系。因為每個連接都有一個相關聯的服務端等待線程，因此可以有很好的響應時間。然而，高負載需要更多的同步運行的線程，這些限制了可擴展性。尤其是，長時間存活的連接像持久化的HTTP連接導致大量的同步工作線程存在，有浪費時間等待新的客戶端請求的趨勢。此外，成百上千的同步線程會浪費大量的棧空間。注意，舉例來說，Solaris/Sparc默認的JAVA棧空間是512KB.

如果server不得不處理大量同時發生的客戶端，并且能容忍慢，無反應的客戶端，就需要一種供替代的線程架構。每個事件一個線程的方式通過一種非常高效地方式實現了這樣的需求。工作線程和連接獨立，僅被用來處理特定的事件。舉例來說，如果一個數據接收事件發生了，一個工作線程將會用來處理特定于應用程序的編碼和服務任務（或至少啟動這些任務）。任務一結束，工作線程就會回到線程池中。這種方式需要無阻塞的處理socket的I/O。調用socket的read或write方法需要時無阻塞的。此外，一個事件系統是必須的；它會發信號表明是否有新數據，輪流發起socket的read方法。這種方式移除了等待線程和工作線程之間的一對一關系。這樣一個事件驅動的I/0系統的設計將會在反應堆模式中描述。

The Reactor Pattern反應堆模式

反應堆模式，如圖1所示，把事件的檢測例如準備就緒讀或者準備就緒接受數據和事件的處理分離。如果一個準備就緒的事件發生了，專用工作線程內的一個事件處理器就會被通知去執行適當的處理。

Figure 1. A NIO-based Reactor pattern implementation

連接通道需要先在Selector類中注冊才能參與事件的架構。這可以通過調用regisster()方法來實現。雖然這個方法是SocketChannel的一部分，這個通道將會在Selector中注冊，沒有其它的方法。

...
SocketChannel channel = serverChannel.accept();
channel.configureBlocking(false);

// register the connection
SelectionKey sk = channel.register(selector, SelectionKey.OP_READ);
...

為了檢測新的事件，Selector類提供了請求已注冊的通道就緒事件的能力。通過調用select方法，Selector收集已注冊通道的就緒事件。這個方法的調用會阻塞，直到至少一個事件已經發生。在這種情況下，方法返回了自上次調用之后就緒的I/O操作的連接數。所選的連接可以通過調用Selector的selectedkey方法來檢測。這個方法返回一個Selectionkey對象集合，里面存放了IO事件的狀態和連接通道的引用。

一個Selector存在于Dispatcher中。這是一個單線程的活動類圍饒著Selector類。Dispatcher類的職責是檢測事件然后分發消費事件的處理給EventHandler類。在這個分發循環中，Dispatcher類調用Selector類的select方法等待新的事件。如果至少一個事件發生了，這個方法就返回，每個事件相關的通道可以通過調用selectedkeys方法獲得。

...
while (isRunning) {
  // blocking call, to wait for new readiness events
  int eventCount = selector.select(); 

  // get the events
  Iterator&lt;SelectionKey&gt; it = selector.selectedKeys().iterator();
  while (it.hasNext()) {
    SelectionKey key = it.next();
    it.remove();

    // readable event?
    if (key.isValid() &amp;&amp; key.isReadable()) {
      eventHandler.onReadableEvent(key.channel());
    }

    // writable event?
    if (key.isValid() &amp;&amp; key.isWritable()) {
      key.interestOps(SelectionKey.OP_READ); // reset to read only
      eventHandler.onWriteableEvent(key.channel());
    }
    ...
  }
  ...
}

基于一個事件，類似于就緒讀或就緒寫，EventHandler會被Dispatcher調用來處理這個事件。EventHandler解碼請求數據，處理必須的服務活動，編碼響應數據。由于工作線程沒有被強制去浪費時間等待新的請求然后建立一個連接，這種方式的可擴展性和吞吐量理論上只限制于系統資源像CPU和內存。這既便是說，響應時間將沒有每個連接一個線程的方式快，由于參與線程間的切換和同步。事件驅動方法的挑戰因此是最少化同步和優化線程管理，以致于這些影響可以被忽略。

Component Architecture組件架構

大多數具有高可擴展性的JAVA服務器都是建立在反應堆模式上的。這樣做，反應堆模式中的類將會被增強，因為需要額外的類來連接管理，緩沖區管理，以及負載均衡。這個服用器的入口類是一個Acceptor。這個安排如圖2所示。

Figure 2. Major components of a connection-oriented server

Acceptor接收器

一個服務器每個新的客戶端連接將會被單個Acceptor所接收，Acceptor與服務器的端口綁定。接收器是一個單線程的活動類。由于Acceptor僅負責處理歷時非常短的客戶端連接請求，經常只要用阻塞I/0模式實現Acceptor就足夠了。Acceptor通過調用Serversocketchannel的阻塞accept方法來處理新請求。新請求將會注冊到Dispatcher，這之后，請求就可以參與到事件處理中了。

由于一個Dispatcher的可擴展性非常有限，通常都會使用一個小的Dispatchers的池。這個限制當中的一個原因是特定的操作系統實現的Selector。大多數的操作系統一對一的映射SocketChannel和文件處理。取決于具體的系統，每個Selector的最大文件處理數的限制也是不同的。

class Acceptor implements Runnable {
  ...
  void init() {
    ServerSocketChannel serverChannel = ServerSocketChannel.open();
    serverChannel.configureBlocking(true);
    serverChannel.socket().bind(new InetSocketAddress(serverPort));
  }

  public void run() {
    while (isRunning) {
      try {
        SocketChannel channel = serverChannel.accept(); 

        Connection con = new Connection(channel, appHandler);
        dispatcherPool.nextDispatcher().register(con);
      } catch (...) {
        ...
      }
    }
  }
}

在示例代碼中，一個連接對象持有SocketChannel和應用級別的事件處理器。我們將會在下面描述這些類。

Dispatcher分配器

通過調用Dispatcher的register方法，SocketChannel將會注冊到相關的Selector上。這里就是問題的來源。Selector在內部使用key集合來管理注冊的通道。這意味著每次注冊一個通道，一個相關連的SelectionKey會被創建并被加入到Selector的注冊key集合。同時，并發的分發線程可以調用Selector的select方法，也會訪問這個key集合。由于key集合是非線程安全的，一個非同步的Acceptor上下文注冊會導致死鎖和競爭。這個可以通過實現selector guard object idiom來解決，它允許暫時的掛起分配線程。參考”“http://developers.sun.com/learning/javaoneonline/2006/coreplatform/TS-1315.pdf”> How to Build a Scalable Multiplexed Server with NIO” (PDF)來查看這個方法的解釋。

 class Dispatcher implements Runnable {
  private Object guard = new Object();
  &hellip;

  void register(Connection con) {
    // retrieve the guard lock and wake up the dispatcher thread
    // to register the connection's channel
    synchronized (guard) {
      selector.wakeup();
      con.getChannel().register(selector, SelectionKey.OP_READ, con);
    }

    // notify the application EventHandler about the new connection
    &hellip;
  }

  void announceWriteNeed(Connection con) {
    SelectionKey key = con.getChannel().keyFor(selector);
    synchronized (guard) {
      selector.wakeup();
      key.interestOps(SelectionKey.OP_READ | SelectionKey.OP_WRITE);
    }
  }

  public void run() {
    while (isRunning) {
      synchronized (guard) {
        // suspend the dispatcher thead if guard is locked
      }
      int eventCount = selector.select();

      Iterator&lt;SelectionKey&gt; it = selector.selectedKeys().iterator();
      while (it.hasNext()) {
        SelectionKey key = it.next();
        it.remove();

        // read event?
        if (key.isValid() &amp;&amp; key.isReadable()) {
          Connection con = (Connection) key.attachment();
          disptacherEventHandler.onReadableEvent(con);
        }

        // write event?
        &hellip;
      }
    }
  }
}

在這個連接注冊之后，Selector監聽這個連接的就緒事件。如果一個事件發生了，通過傳遞相關的連接，這個Dispatcher的事件處理類的合適的回調方法將會被調用。

分配器級別事件處理器

處理一個就緒讀事件的第一個行為是調用通道的讀方法。與流接口相反，通道接口需要忽略讀緩沖接口。通常會使用直接分配的ByteBuffer。直接緩沖區存在于本地內存，繞過JAVA堆內存。通過使用直接緩沖，socket的IO操作不再需要創建內部中間緩沖器。

通常情況下，讀請求會被非常快的執行。Socket的讀操作通常只是把一份接收到的數據從內核內存空間拷貝到讀緩沖區，這個數據會存在于用戶控制的內存空間。這些接收的數據將會被添加到連接的線程安全的讀隊列作進一步的處理。基于I/O操作的結果，特定于應用程序的任務會被執行。這些任務會被分配的應用級別的事件處理器處理。這類處理器通常被稱為工作線程。

 class DispatcherEventHandler {
  ...

  void onReadableEvent(final Connection con) {
    // get the received data
    ByteBuffer readBuffer = allocateMemory();
    con.getChannel().read(readBuffer);
    ByteBuffer data = extractReadAndRecycleRenaming(readBuffer);

    // append it to read queue
    con.getReadQueue().add(data);
    ...

    // perform further operations (encode, process, decode)
    // by a worker thread
    if (con.getReadQueue().getSize() &gt; 0) {
      workerPool.execute(new Runnable() {
        public void run() {
          synchronized (con) {
            con.getAppHandler().onData(con);
          }
        }
      });
    }
  }

  void onWriteableEvent(Connection con) {
    ByteBuffer[] data = con.getWriteQueue().drain();
    con.getChannel().write(data); // write the data
    ...

    if (con.getWriteQueue().isEmpty()) {
      if (con.isClosed()) {
        dispatcher.deregister(con);
      }

    } else {
       // there is remaining data to write
       dispatcher.announceWriteNeed(con);
    }
  }
}

在特定于應用程序的任務中，數據會被編碼，服務會被執行，數據會被寫入。在寫數據的時候，要被發送的數據會加入到寫隊列，然后調用Dispatcher類的announceWriteNeed方法。這個方法讓Selector開始監聽就緒讀事件。如果這種事件發生，分配器級別的事件處理器就會執行onWriteableEvent方法。這從通道的寫隊列獲取數據然后執行必要的寫I/O操作。試圖直接寫數據，通過這種方法，將會導致死鎖和競爭。

應用級別事件處理器

與分配器事件處理器相比，特定于應用的事件處理器監聽高級別的面向連接的事件，例如建立連接，數據接收或者是關閉連接。具體的事件處理設計是NIO服務器框架像SEDA，MINA還有emberIO之間最大的不同。這些框架通常實現了多級的架構，這樣事件處理鏈就可以使用。它允許增加像SSLHandler或DelayerWriteHandler之類可以攔截請求/響應處理的處理器。下面的例子展示了一個基于xSocket框架的應用級別的處理器。xScoket框架支持不同的處理器接口，這些接口里面定義了需要被實現的特定于應用的回調方法代碼。

 class POP3ProtocolHandler implements IConnectHandler, IDataHandler, ... {
  private static final String DELIMITER = ...
  private Mailbox mailbox = ...

  public static void main(String... args) throws ... {
    new MultithreadedServer(110, new POP3ProtocolHandler()).run();
  }

  public boolean onConnect(INonBlockingConnection con) throws ... {
    if (gatekeeper.isSuspiciousAddress(con.getRemoteAddress())) {
      con.setWriteTransferRate(5);  // reduce transfer: 5byte/sec
    }

    con.write("+OK My POP3-Server" + DELIMITER);
    return true;
  }

  public boolean onData(INonBlockingConnection con) throws ... {
    String request = con.readStringByDelimiter(DELIMITER);

    if (request.startsWith("QUIT")) {
      mailbox.close();
      con.write("+OK POP3 server signing off" + DELIMITER);
      con.close();

    } else if (request.startsWith("USER")) {
      this.user = request.substring(4, request.length());
      con.write("+OK enter password" + DELIMITER);

    } else if (request.startsWith("PASS")) {
      String pwd = request.substring(4, request.length());
      boolean isAuthenticated = authenticator.check(user, pwd);
      if (isAuthenticated) {
        mailbox = MailBox.openAndLock(user);
        con.write("+OK mailbox locked and ready" + DELIMITER);
      } else {
        ...
      }
    } else if (...) {
      ...
    }
    return true;
  }
}

為了更簡便的訪問底層的讀寫隊列，Connection對象提供了一些便利的面向流和通道的讀寫方法。

通過關閉連接，底層實現初始化一個可寫事件往返的刷新寫隊列。連接會在遺留的數據被寫完之后終止。除了這樣一個控制終端，連接還能因為其它的原因關閉。例如，硬件故障可能導致基于TCP的連接中斷。這樣的情況只有在socket上執行讀寫操作或空閑超時的時候檢測到。大多數的NIO框架提供一個內置的程序來處理這些不受控制的中斷。

Conclusion總結

一個事件驅動的非阻塞架構是實現高效，高擴展性和高穩定性服務器的一個基本的層。其中的挑戰就是最小化線程同步開銷和優化連接和緩沖區的管理。這會是編程中最困難的部分。

但是沒有必要重復發明輪子。一些框架像xSocket，emberIO，SEDA或MINA都抽象了低層次的事件處理和線程管理來簡化創建高可擴展性的服務器。以上大部分的框架都支持SSL和UDP，本文中未提及這兩點。

Resources參考資料

Scalable IO in Java” (PDF) describes event-driven processing by using Java NIO
“Tricks and Tips with NIO, Part 2: Why SelectionKey.attach() Is Evil” describes how a memory leak occurs by a unwary use of the
SelectionKey’s attach
“Pico Threads: Lightweight Threads in Java” shows the problems with
large-scale threaded programming and event-based techniques.
A Reactor pattern description by Douglas C. Schmidt (PDF)
Unix Network Programming: The Sockets Networking API gives a good overview about network programming in general, and gives a good impression what happens behind the Java I/O operations on the operating-system level.
xSocket is a LGPL NIO-based library to build network applications. Most example code of this article has been written based on xSocket.

作者：

Gregor Roth works as a software architect at United Internet group, a leading European Internet Service Provider to which GMX, 1&1, and Web.de belong. His areas of interest include software and system architecture, enterprise architecture management, object-oriented design, distributed computing, and development methodologies.

來自：http://ifeve.com/a-scalable-nonblocking-io-server/

本文由用戶 jopen 自行上傳分享，僅供網友學習交流。所有權歸原作者，若您的權利被侵害，請聯系管理員。

轉載本站原創文章，請注明出處，并保留原始鏈接、圖片水印。

本站是一個以用戶分享為主的開源技術平臺，歡迎各類分享！

本文地址：http://www.baiduhome.net/lib/view/open1415257560902.html

架構 Java開發