程式師世界 >> 編程語言 >> JAVA編程 >> 關於JAVA >> 構建服務器集群感知的Java應用程序

構建服務器集群感知的Java應用程序

編輯：關於JAVA

如今，許多企業應用程序都由一組合作的分布式進程和服務器交付。例如，可向幾乎所有流行的 Java 企業服務器的 Web 請求提供服務器集群功能，這些服務器還可以提供有限的配置選項，如服務器權重和配置重新加載。

雖然大多數 Java 企業服務器具有集群的內置支持，但對於自定義用例來說，在應用程序級並沒有現成提供這種支持。作為軟件開發人員，我們應該如何管理涉及分布式任務協調或支持多租戶應用程序的用例？（多租戶應用程序是要求實例在整體服務器集群或組的子集上被隔離的應用程序。）對於這些類型的用例，我們必須找到一種方法，使得組協調功能在應用程序軟件層上可用，最好能在高級別的抽象層上提供。

在本文中，我們提供了將組成員和管理功能融合到分布式 Java 應用程序中的指南。我們將從一個基於 Spring Integration 的模擬 Java 應用程序開始，利用基於兩個開源項目（Apache ZooKeeper 和 LinkedIn 的 Project Norbert）的服務器集群抽象層來進行構建。

服務器集群概述

服務器集群感知的應用程序通常至少需要以下功能：

具有狀態維護和查詢功能的組成員：需要實時組成員，以便在一組活動的服務器上分發處理。為了管理組成員，應用程序必須能夠建立一個進程/服務器組，並跟蹤該組中所有服務器的狀態。當某台服務器停機或上線時，它還必須能夠通知活動的服務器。該應用程序將只在集群中的活動服務器之間對服務請求執行路由和負載均衡，從而幫助確保實現高可用的服務。

主要進程或領袖進程：這是一個在集群中的進程，負責維護整個服務器集群的同步狀態的協調功能。選擇領袖進程的機制是被稱為分布式共識的一組更廣泛的問題中的一個特例。（兩階段提交和三階段提交是眾所周知的分布式共識問題。）

任務協調和動態的領袖服務器選舉：在應用程序級別，領袖服務器負責任務協調，通過集群中的其他（跟隨者）服務器之間分發任務來做到這一點。擁有一台領袖服務器，可以消除服務器之間潛在的爭用，否則爭用將需要某種形式的互斥或鎖定，才可以使符合資格的任務運行（例如，服務器對來自公共數據存儲的任務進行輪詢）。正是動態領袖選舉使得分布式處理變得可靠；如果領袖服務器崩潰，可以選舉新的領袖繼續處理應用程序任務。

組通信：在一個集群感知的應用程序中的應用程序應該能夠在整個服務器集群中促進結構化數據和命令的高效交換。

分布式鎖和共享數據：如果有需要，分布式應用程序應該能夠訪問的特性包括分布式鎖和共享數據結構，如隊列和映射。

示例：Spring Integration

我們的代表示例是一個企業應用程序集成 (EAI) 場景，我們將利用基於 Spring Integration 的模擬應用程序處理該場景。該應用程序具有以下特征和要求：

一個模擬源應用程序產生與集成相關的事件和消息作為其日常事務處理的一部分，並將它們存儲在一個數據存儲中。

集成事件和消息由一組分布式 Java 進程（一個服務器集群）進行處理，這些進程可以在同一台服務器上運行，也可以分布在由一個高性能網絡連接的多台計算機上。需要服務器集群來實現可擴展性和高可用性。

每個集成事件只由任一集群成員（即特定的 JVM）處理一次。輸出消息通過 Intranet 或 Internet 被傳遞給合作伙伴應用程序（如果適用）。

圖 1 顯示了集成事件和從模擬源應用程序出站的消息處理流。

圖 1. 基於 Spring Integration 的示例應用程序示意圖

設計解決方案的架構

要為該用例開發一個解決方案，第一步是分發要在一個服務器集群上運行的集成應用程序。這應該能增加處理吞吐量，並能確保高可用性和可擴展性。單次進程的失敗不會中止整個應用程序。

一旦完成分配，集成應用程序將從應用程序的數據存儲中獲得集成事件。服務器集群中的單台服務器將通過一個合適的應用程序適配器從事件存儲獲取應用程序事件，然後將這些事件分發給集群中的其余服務器進行處理。這個單台服務器擔任領袖服務器，或任務協調者的角色，負責將事件（處理任務）分發給整個集群中的其余服務器。

支持集成應用程序的服務器集群成員在配置時已眾所周知。每當有新服務器啟動，或有任何服務器崩潰或停機時，集群成員信息就會動態分發給所有運行的服務器。同樣，任務協調者服務器也是動態選擇的，如果任務協調者服務器崩潰或變得不可用，將從其余運行中的服務器中以合作方式選一個備用的領袖服務器。集成事件可能由支持企業集成模式 (Enterprise Integration Patterns, EIP) 的多個開源 Java 框架其中之一處理。

圖 2 顯示了用例解決方案的示意圖和組件，我們會在下一節中進一步描述它們。

圖 2. 用例解決方案的示意圖和服務器集群組件

服務器集群

我們的集成應用程序需要服務器組相關的特性，但是無論是 Java 標准版 (Java SE) 還是 Java 企業版 (Java EE) 均沒有現成提供這些特性。這些示例包括服務器集群和動態服務器領袖選舉。

圖 3 顯示了我們將用於實施我們的 EAI 解決方案的開源工具，即 Spring Integration 實現事件處理，以及 Apache ZooKeeper 和 LinkedIn Project Norbert 實現集群感知。

圖 3. 服務器集群的技術映射

關於模擬的應用程序

模擬應用程序的目的是演示如何使用 Apache ZooKeeper 和 Project Norbert 來解決在開發基於 Java 的服務器集群中所面臨的常見挑戰。該應用程序的工作原理如下：

應用程序事件存儲由一個共享文件夾來模擬，集成服務器集群內的所有服務器都可以訪問該文件夾。

使用保存在此共享文件夾上的文件（以及其數據）來模擬集成事件。也可以使用外部腳本來不斷創建文件，從而模擬事件的創建。

基於 Spring Integration 的文件輪詢組件（入站事件適配器）從應用程序事件存儲獲取事件，這是由共享文件系統文件夾模擬的事件。然後，文件數據被分發到其余的服務器集群成員進行處理。

事件處理的模擬方法是：用簡短的標頭類型信息（比如 server id 和 timestamp）作為文件數據的前綴。

合作伙伴應用程序由其他一些獨特的共享文件系統模擬文件夾，每個合作伙伴應用程序對應一個文件夾。

現在已概述過示例用例、建議的解決方案架構，以及模擬的應用程序。現在，我們准備介紹服務器集群和任務分發解決方案的兩個主要組件：Apache ZooKeeper 和 LinkedIn 的 Project Norbert。

Apache ZooKeeper 和 Project Norbert

首先由 Yahoo Research 開發的 ZooKeeper 最初被 Apache Software Foundation 采納為 Hadoop 的一個子項目。如今，它已是一個頂級項目，可提供分布式組協調服務。我們將使用 ZooKeeper 創建服務器集群來托管我們的模擬應用程序。ZooKeeper 也將實現應用程序所需要的服務器領袖選舉功能。（領袖選舉對於 ZooKeeper 提供的所有其他組協調功能是必不可少的。）

ZooKeeper 服務器通過 znode（ZooKeeper 數據節點）實現服務器協調，znode 是在內存中復制的類似於分層文件系統的數據模型。和文件一樣，znode 可以保存數據，但它們也像目錄一樣，可以包含子級 znode。

有兩種類型的 znode：常規 znode 由客戶端進程顯式創建和刪除，而暫時性 znode 在發起的客戶端會話停止時會自動被刪除。若常規或暫時性 znode 在創建時帶有順序標志，一個 10 位數字的單調遞增後綴將被附加到 znode 名稱。

關於 ZooKeeper 的更多信息：

ZooKeeper 確保當服務器啟動時，每台服務器都知道該組中其他服務器的偵聽端口。偵聽端口支持促進領袖選舉、對等通信，以及客戶端與服務器的連接的服務。

ZooKeeper 使用一個組共識算法來選舉領袖，完成選舉之後，其他服務器都稱為跟隨者。服務器集群需要服務器數量達到下限（法定數量）時才可以執行。

客戶端進程有一組已定義的可用操作，他們使用這些操作基於 znode 編排數據模型的讀取和更新。

所有寫操作都通過領袖服務器進行路由，這限制了寫操作的可擴展性。領袖使用稱為 ZooKeeper Atomic Broadcast (ZAB) 的廣播協議來更新跟隨者服務器。ZAB 保留更新順序。因此，在內存中的類似於文件系統的數據模型最終在組或集群中的所有服務器上保持同步。數據模型也通過持久的快照被定期寫入到磁盤。

讀操作的可擴展性比寫操作高得多。跟隨者從數據模型的這個同步副本響應客戶端進程讀取。

znode 支持客戶端的一次性回調機制，被稱為 “看守者”。看守者觸發一個監視客戶端進程的信號，監視被看守的 znode 的更改。

利用 Project Norbert 實現組管理

LinkedIn 的 Project Norbert 掛接到一個基於 Apache ZooKeeper 的集群，以提供具有服務器集群成員感知的應用程序。Norbert 在運行時動態完成該操作。Norbert 也封裝了一個 JBoss Netty 服務器，並提供了相應的客戶端模塊來支持應用程序交換消息。需要注意的是，早期版本的 Norbert 需要使用 Google Protocol Buffers 對象序列化消息庫。目前的版本支持自定義對象的序列化。（請參閱參考資料了解更多信息。）

Spring Integration

Spring Integration 是一個解決 EAI 挑戰的開源框架。它通過聲明組件支持 Enterprise Integration Patterns（企業集成模式），並建立在 Spring 編程模型之上。

構建一個服務器集群

准備好所有組件之後，我們就可以開始配置事件處理服務器集群了。配置集群的第一步是建立服務器法定數量，之後，新當選的領袖服務器會自動啟動本地文件輪詢流。文件輪詢通過 Spring Integration 發生，Spring Integration 模擬一個入站應用程序事件適配器。使用一個循環任務分配策略，將輪詢過的文件（模擬應用程序事件）分發到可用的服務器中。

需要注意的是，ZooKeeper 將有效的法定數量定義為服務器進程的大多數。因此，一個集群至少由三個服務器組成，在至少兩個服務器處於活動狀態時建立法定數量。此外，在我們的模擬應用程序中，每台服務器都需要兩個配置文件：一個屬性文件，由引導整個服務器 JVM 的驅動程序使用，還有一個單獨的屬性文件供基於 ZooKeeper 的服務器集群（在該集群中，每個服務器都是一個部分）使用。

第 1 步：創建一個屬性文件

Server.java是一個控制器和條目類，用來啟動我們的分布式 EAI 應用程序。應用程序的初始參數是從屬性文件中讀取的，如清單 1 所示：

清單 1. 服務器屬性文件

# Each server in group gets a unique id:integer in range 1-255
server.id=1
    
# zookeeper server configuration parameter file -relative path to this bootstrap file
zookeeperConfigFile=server1.cfg
    
#directory where input events for processing are polled for - common for all servers
inputEventsDir=C:/test/in_events
    
#directory where output / processed events are written out - may or may not be shared by 
#all servers
outputEventsDir=C:/test/out_events
    
#listener port for Netty server (used for intra server message exchange)
messageServerPort=2195

注意，在這個最小的服務器集群中，每一台服務器都需要一個惟一的 server id（整數值）。

輸入事件目錄被所有服務器共享。輸出事件目錄模擬一個合作伙伴應用程序，並可以視情況由所有服務器共享。ZooKeeper 分發提供了一個類，用於解析服務器集群的每個成員服務器或 “法定數量對等服務器” 的配置信息。因為我們的應用程序重用了這個類，所以它需要相同格式的 ZooKeeper 配置。

還需要注意的是，messageServerPort 是 Netty 服務器（由 Norbert 庫啟動和管理）的偵聽器端口。

第 2 步：為進程中的 ZooKeeper 服務器創建一個配置文件

清單 2. ZooKeeper 的配置文件 (server1.cfg)

tickTime=2000
dataDir=C:/test/node1/data
dataLogDir=C:/test/node1/log
clientPort=2181
initLimit=5
syncLimit=2
peerType=participant
maxSessionTimeout=60000
server.1=127.0.0.1:2888:3888
server.2=127.0.0.1:2889:3889
server.3=127.0.0.1:2890:3890

ZooKeeper 文檔中介紹了在清單 2 中顯示的參數（以及一些可選參數，這些參數均使用默認值，覆蓋的參數除外）。需要注意的是，每個 ZooKeeper 服務器使用三個偵聽器端口。clientPort（在上面的配置文件中是 2181）由連接到服務器的客戶端進程使用；第二個偵聽器端口用於實現對等通信（對於服務器 1 的值是 2888）；第三個偵聽器端口支持領袖選舉協議（對於服務器 1 的值是 3888）。每台服務器都支持集群的整體服務器拓撲結構，所以 server1.cfg 也列出了服務器 2 和服務器 3 及其對等端口。

第 3 步：在服務器啟動時初始化 ZooKeeper 集群

控制器類 Server.java 啟動一個單獨的的線程 (ZooKeeperServer.java)，該線程封裝基於 ZooKeeper 的集群成員，如清單 3 所示：

清單 3. ZooKeeperServer.java

package 

ibm.developerworks.article;
…
public class ZooKeeperServer implements Runnable
{
    
   public ZooKeeperServer(File configFile) throws ConfigException, IOException
   {
      serverConfig = new QuorumPeerConfig();
      …
      serverConfig.parse(configFile.getCanonicalPath());
    }
    
      public void run()
   {
      NIOServerCnxn.Factory cnxnFactory;
      try
      {
         // supports client connections
         cnxnFactory = new NIOServerCnxn.Factory(serverConfig.getClientPortAddress(),
               serverConfig.getMaxClientCnxns());
         server = new QuorumPeer();
    
         // most properties defaulted from QuorumPeerConfig; can be overridden
         // by specifying in the zookeeper config file
    
         server.setClientPortAddress(serverConfig.getClientPortAddress());
         …
         server.start(); //start this cluster member
    
         // wait for server thread to die
         server.join();
      }
      …
   }
    
    …
   public boolean isLeader()
   {
      //used to control file poller.  Only the leader process does task
      // distribution
      if (server != null)
      {
         return (server.leader != null);
      }
      return false;
   }

第 4 步：初始化基於 Norbert 的消息傳送服務器

建立了服務器法定數量後，我們可以啟動基於 Norbert 的 Netty 服務器，該服務器支持快速的服務器內部消息傳送。

清單 4. MessagingServer.java

public static void init(QuorumPeerConfig config) throws UnknownHostException
{
    …
   // [a] client (wrapper) for zookeeper server - points to local / in process
   // zookeeper server
   String host = "localhost" + ":" + config.getClientPortAddress().getPort();
    
   //[a1] the zookeeper session timeout (5000 ms) affects how fast cluster topology 
   // changes are communicated back to the cluster state listener class
    
   zkClusterClient = new ZooKeeperClusterClient("eai_sample_service", host, 5000);
    
   zkClusterClient.awaitConnectionUninterruptibly();
   …
   // [b] nettyServerURL - is URL for local Netty server URL
   nettyServerURL = String.format("%s:%d", InetAddress.getLocalHost().getHostName(),
         Server.getNettyServerPort());
   …
    
   // [c]
   …
   zkClusterClient.addNode(nodeId, nettyServerURL);
    
   // [d] add cluster listener to monitor state
   zkClusterClient.addListener(new ClusterStateListener());
    
   //  Norbert - Netty server config
   NetworkServerConfig norbertServerConfig = new NetworkServerConfig();
    
   // [e] group coordination via zookeeper cluster client
   norbertServerConfig.setClusterClient(zkClusterClient);
    
   // [f] threads required for processing requests
   norbertServerConfig.setRequestThreadMaxPoolSize(20);
    
   networkServer = new NettyNetworkServer(norbertServerConfig);
    
   // [g] register message handler (identifies request and response types) and the
   // corresponding object serializer for the request and response
   networkServer.registerHandler(new AppMessageHandler(), new CommonSerializer());
    
   // bind the server to the unique server id
   // one to one association between zookeeper server and Norbert server
   networkServer.bind(Server.getServerId());   
    
}

請注意，基於 Norbert 的消息傳送服務器包括一個連接到 ZooKeeper 集群的客戶端。配置此服務器，連接到本地（進程中）的 ZooKeeper 服務器，然後為 ZooKeeper 服務器創建一個客戶端。會話超時將影響集群的拓撲結構更改可以多快傳回應用程序。這將有效地創建一個較小的時間窗口，在該時間窗口內，集群拓撲結構的記錄狀態將與集群拓撲結構的實際狀態不同步，這是因為新的服務器啟動或現有的服務器崩潰。應用程序需要在這段時間內緩沖消息或實施消息發送失敗的重試邏輯。

MessagingServer.java (清單 4) 執行以下任務：

為 Netty 服務器配置端點 (URL)。

將本地 node Id 或 server Id 與已配置的 Netty 服務器相關聯。

關聯一個集群狀態偵聽器（我們將會簡略介紹）實例。Norbert 將使用該實例將集群拓撲結構更改推送回應用程序。

將 ZooKeeper 集群客戶端分配給正在進行填充的服務器配置實例。

為請求/響應對關聯一個惟一的消息處理程序類。也需要一個序列化程序類來對請求和響應對象進行編組和解組。

還要注意的是，消息傳送的應用程序回調需要一個線程池。

第 5 步：初始化 Norbert 集群客戶端

接下來，初始化 Norbert 集群客戶端。MessagingClient.java（如清單 5 所示）配置集群客戶端，並使用一個負載均衡策略初始化該客戶端：

清單 5. MessagingClient.java

public class MessagingClient
{
   …
   public static void init()
   {
      …
      NetworkClientConfig config = new NetworkClientConfig();
    
      // [a] need instance of local norbert based zookeeper cluster client
      config.setClusterClient(MessagingServer.getZooKeeperClusterClient());
    
      // [b] cluster client with round robin selection of message target
      nettyClient = new NettyNetworkClient(config, 
                                 new RoundRobinLoadBalancerFactory());
      …
   }
   ...
    …
   // [c] – if server id <0 – used round robin strategy to choose target
   // else send to identified target server 
   public static Future<String> sendMessage(AppRequestMsg messg, int serverId)
         throws Exception
   {
      …
      // [d] load balance message to cluster- strategy registered with client
      if (serverId <= 0)
      {
         …
         return nettyClient.sendRequest(messg, serializer);
      }
      else
      {
         // [e] message to be sent to particular server node
         …
         if (destNode != null)
         {
            …
            return nettyClient.sendRequestToNode(messg, destNode, serializer);
         }
         …
      }
   }
   …
}

注意，在清單 5 中，如果沒有由一個正的 server Id 值識別目標服務器，根據已配置的負載均衡策略從活動的組中選擇服務器會發出消息。應用程序可以配置和實現它們自己的消息處理策略，也許以其他服務器屬性為依據。

狀態監測和任務分發

模擬的應用程序還有三個組件，我們將在下面的章節中進行介紹：

一個組件用於監視集群的狀態（服務器成員）。

一個 Spring Integration 流定義文件。進程定義文件定義基於 EIP 的消息流，包括從模擬應用程序的任務池流到中心任務分發程序。任務分發程序最終將每個任務路由到其中一個可用的集群成員，以進行處理。

一個任務分發程序，該程序實施最終的任務路由，將任務路由到其中一個集群成員。

集群狀態（拓撲結構）偵聽器

集群狀態偵聽器確保消息傳送客戶端擁有最新的可用節點列表。該偵聽器也在領袖服務器上啟動惟一的事件適配器實例（文件輪詢程序）。文件輪詢程序將輪詢過的文件移交給消息處理器組件（一個 POJO），這就是實際的任務分發程序。由於集群內只有一個任務分發程序實例，所以不需要進一步的應用程序級同步。清單 6 顯示了集群狀態偵聽器：

清單 6. ClusterStateListener.java

public 

class ClusterStateListener implements ClusterListener
{
   …
   public void handleClusterNodesChanged(Set<Node> currNodeSet)
   {
      …
      // [a] start event adapter if this server is leader
      if (Server.isLeader() && !Server.isFilePollerRunning())
      {
         Server.startFilePoller();
      }
    
   }
   …
}

基於 Spring Integration 的文件輪詢程序

Spring Integration 流執行以下任務（如清單 7 所示）：

創建一條稱為 messageInputChannel 的消息通道或管道。

定義一個入站通道適配器，每隔 10 秒輪詢一次從 JVM 系統屬性讀取的目錄（即 property input.dir）中的文件。被輪詢和定位的任何文件都被向下傳遞給消息通道 messageInputChannel。

配置任務分發程序 Java Bean，用於從消息通道接收消息。調用其方法 processMessage 來運行任務分發函數。

清單 7. 基於 Spring Integration 的流：FilePoller_spring.xml

…
   <integration:channel id="messageInputChannel" />
    
    <int-file:inbound-channel-adapter channel="messageInputChannel"
        directory="file:#{ systemProperties['input.dir'] }"
        filename-pattern="*.*" prevent-duplicates="true" >
    
        <integration:poller id="poller" fixed-rate="10" />
    </int-file:inbound-channel-adapter>
    
    <integration:service-activator input-channel="messageInputChannel"
        method="processMessage" ref="taskDistributor" />
    
    <bean
        id="taskDistributor"
        class="ibm.developerworks.article.TaskDistributor" >
    </bean>
…

任務分發程序

任務分發程序包含跨集群成員路由請求的邏輯。文件輪詢程序組件僅在領袖服務器上激活，並且將輪詢過的文件（在本例中為模擬的集成事件）傳遞到任務分發程序。任務分發程序使用 Norbert 客戶端模塊，將請求（即封裝成消息的輪詢過的文件）路由到集群中的活動服務器。任務分發程序如清單 8 所示：

清單 8. Spring Integration 流控制的任務分發程序（一個 Java Bean）

{
   …
   // [a] invoked by spring integration context 
   public void processMessage(Message<File> polledMessg)
   {
      File polledFile = polledMessg.getPayload();
      …
          
      try
      {
         logr.info("Received file as input:" + polledFile.getCanonicalPath());
    
         // prefix file name and a delimiter for the rest of the payload
         payload = polledFile.getName() + "|~" + Files.toString(polledFile, charset);
    
         …
         // create new message
         AppRequestMsg newMessg = new AppRequestMsg(payload);
    
         // [b]load balance the request to operating servers without
         // targeting any one in particular
         Future<String> retAck = MessagingClient.sendMessage(newMessg, -1);
    
         // block for acknowledgement - could have processed acknowledgement
         // asynchronously by repositing to a separate queue
         String ack = retAck.get();
         …
         logr.info("sent message and received acknowledgement:" + ack);
      …
   }
}

請注意，服務激活器方法的調用是通過在文件輪詢程序找到一個文件進行處理之後控制 Spring Integration 上下文來完成。另外請注意，文件的內容是序列化的，並形成一個新請求對象的有效負載。消息傳送客戶端的 sendMessage() 方法被調用，但沒有針對某一特定的目標服務器。然後，Norbert 客戶端模塊將結果消息負載均衡到集群的其中一台運行的服務器。

運行模擬的應用程序

一個 “可運行” 的 JAR 文件 (ZooKeeper-Norbert- simulated-app.jar) 與三個服務器集群的樣例配置文件都包含在本文的源代碼中。

若要測試應用程序，您可以在本地的單台計算機上啟動所有三個服務器，或在整個網絡中分發它們。為了在多台計算機上運行應用程序，需要一個可從網絡安裝/訪問的公共輸入事件文件夾。通過創建相應的配置文件，每增加一台服務器就創建兩個配置文件，並更新現有的配置文件之後，您就可以在集群中增加服務器的數量。

觸發器事件處理，將包含文本內容的文件復制到指定的輸入文件夾。連續文件由不同的服務器進行處理。通過停止其中一台服務器可測試服務的可靠性。（請注意，由三個服務器組成的集群的法定數量規定，在任何時候都只能有一台服務器停機，以保持應用程序正常運行）。默認情況下，所包含的 log4j.properties 文件啟用 TRACE 級的日志記錄；請注意，服務器拓撲結構將隨正在運行的服務器更新。如果您讓領袖服務器停機，那麼將選出新的領袖，並在那台服務器上激活文件輪詢流，從而保證進行連續的處理。

關於JAVA