您现在的位置：程式碼之家 >> 編程語言 > >> Java 程式設計 >> Java 相關知識

構建服務器集群感知的 Java 應用：基於 ZooKeeper 與 Norbert 的分布式協調實戰

編輯：Java 相關知識

為什麼應用層需要集群感知？

Java EE 服務器雖內建集群，但僅限於標準服務（Session 複製、負載均衡）。對於**分布式任務協調**、**多租戶隔離**、**自定義領袖選舉**等業務級需求，必須在應用層引入集群協調能力。本文演示如何整合 Apache ZooKeeper（組成員/領袖選舉）與 LinkedIn Norbert（消息路由/負載均衡），構建一個 Spring Integration 驅動的模擬 EAI 事件處理集群。

架構設計

Spring Integration 應用示意解決方案組件技術映射

模擬事件源：共享文件夾不斷生成文件。
領袖服務器：唯一文件輪詢器，讀取文件分發至集群處理。
跟隨者服務器：接收任務處理，結果寫入輸出目錄。
動態拓撲：節點啟動/掛掉自動感知，領袖失效自動重選。

核心組件詳解

1. Apache ZooKeeper：分布式協調內核

**數據模型**：層級命名空間 znode（類文件系統），支持持久/臨時、有序節點。
**領袖選舉**：基於 ZAB 協議，集群需達多數（Quorum）方可服務，最小 3 節點。
**Watch 機制**：客戶端註冊監聽，節點變更推送通知。
**寫入路由**：所有寫操作經領袖廣播，保證順序一致性；讀操作可從任意 Follower 本地讀取，擴展性強。

2. Project Norbert：應用級集群客戶端

封裝 ZooKeeper 客戶端，提供動態節點發現、會話管理。
內嵌 Netty 服務器/客戶端，支持高性能 RPC 風格消息交換。
可插拔負載均衡策略（輪詢、一致性哈希等）與序列化器（Protobuf、Java 序列化、自定義）。

工程實現步驟

步驟 1：啟動參數屬性文件（每節點唯一）

server.id=1
zookeeperConfigFile=server1.cfg
inputEventsDir=C:/test/in_events
outputEventsDir=C:/test/out_events
messageServerPort=2195

步驟 2：ZooKeeper 配置（server1.cfg）

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

三端口：clientPort 客戶端連接；2888 對等通信；3888 領袖選舉。

步驟 3：進程內啟動 ZooKeeper 服務器

public class ZooKeeperServer implements Runnable {
    private QuorumPeer server;
    public ZooKeeperServer(File configFile) throws ... {
        QuorumPeerConfig config = new QuorumPeerConfig();
        config.parse(configFile.getCanonicalPath());
        this.config = config;
    }
    public void run() {
        NIOServerCnxn.Factory factory = new NIOServerCnxn.Factory(config.getClientPortAddress(), config.getMaxClientCnxns());
        server = new QuorumPeer();
        server.setClientPortAddress(config.getClientPortAddress());
        // ... 其餘屬性從 config 繼承
        server.start(); // 啟動該集群節點
        server.join();  // 阻塞等待
    }
    public boolean isLeader() { return server != null && server.leader != null; }
}

步驟 4：初始化 Norbert 消息服務器

public static void init(QuorumPeerConfig config) {
    String host = "localhost:" + config.getClientPortAddress().getPort();
    ZooKeeperClusterClient zkClient = new ZooKeeperClusterClient("eai_sample_service", host, 5000);
    zkClient.awaitConnectionUninterruptibly();

    String nettyUrl = String.format("%s:%d", InetAddress.getLocalHost().getHostName(), Server.getNettyServerPort());
    zkClient.addNode(Server.getServerId(), nettyUrl); // 註冊自身端點
    zkClient.addListener(new ClusterStateListener()); // 拓撲變更監聽

    NetworkServerConfig nsc = new NetworkServerConfig();
    nsc.setClusterClient(zkClient);
    nsc.setRequestThreadMaxPoolSize(20);
    networkServer = new NettyNetworkServer(nsc);
    networkServer.registerHandler(new AppMessageHandler(), new CommonSerializer());
    networkServer.bind(Server.getServerId()); // 綁定 serverId
}

關鍵點：會話超時（5s）決定拓撲變更感知延遲，應用層需在窗口期內緩衝或重試消息。

步驟 5：Norbert 客戶端與負載均衡

public static void init() {
    NetworkClientConfig cfg = new NetworkClientConfig();
    cfg.setClusterClient(MessagingServer.getZooKeeperClusterClient());
    nettyClient = new NettyNetworkClient(cfg, new RoundRobinLoadBalancerFactory());
}

public static Future<String> sendMessage(AppRequestMsg msg, int serverId) {
    if (serverId <= 0) return nettyClient.sendRequest(msg, serializer); // 輪詢
    else return nettyClient.sendRequestToNode(msg, destNode, serializer); // 指定節點
}

狀態監聽與任務分發

ClusterStateListener：拓撲變更觸發領袖職責

public class ClusterStateListener implements ClusterListener {
    public void handleClusterNodesChanged(Set<Node> nodes) {
        if (Server.isLeader() && !Server.isFilePollerRunning()) {
            Server.startFilePoller(); // 唯一領袖啟動輪詢
        }
    }
}

Spring Integration 文件輪詢流（僅領袖激活）

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

TaskDistributor：輪詢分發任務至集群

public void processMessage(Message<File> polledMsg) {
    File file = polledMsg.getPayload();
    String payload = file.getName() + "|~" + Files.toString(file, Charsets.UTF_8);
    AppRequestMsg msg = new AppRequestMsg(payload);
    Future<String> ack = MessagingClient.sendMessage(msg, -1); // 輪詢分發
    String result = ack.get(); // 同步等待確認，亦可異步回調
    log.info("Sent & ack: " + result);
}

運行與驗證

源碼包含可運行 JAR 與 3 節點配置，單機或分佈式啟動均可。
向輸入目錄投放文本文件觸發處理，不同節點輪流處理（日志顯示 serverId）。
殺掉領袖節點 → 新領袖自動當選 → 輪詢流在新領袖啟動 → 處理不中斷。
日志 TRACE 級記錄拓撲變更、領袖切換、任務分發全過程。

總結與遷移指南

能力	技術選型	適用場景
組成員/狀態	ZooKeeper 臨時節點 + Watch	服務發現、存活檢測、配置推送
領袖選舉	ZooKeeper 有序臨時節點	單點任務調度、分布式鎖、主備切換
集群內 RPC	Norbert + Netty	任務分發、狀態同步、命令下發
負載均衡	Norbert LoadBalancerFactory	無狀態服務路由、多租戶分片
分布式鎖/隊列	ZooKeeper 原語或 Curator 框架	共享資源互斥、有序任務隊列