Zookeeper使用實例——服務節點管理

分布式處理中，總會存在多個服務節點同時工作，並且節點數量會隨着網絡規模的變化而動態增減，服務節點也有可能發生宕機與恢復。面對着動態增減的服務節點，我們如何保證客戶請求被服務器正確處理呢。我們可以通過zookeeper臨時節點創建與自動刪除來掌握服務節點的動態增減。

ignite分布式緩存支持使用zookeeper發現ignite節點的增減，這正是zookeeper管理服務節點的一個典型應用場景。我們來看看關鍵代碼

        // 關鍵方法，創建包含自增長id名稱的目錄，這個方法支持了分布式鎖的實現
        // 四個參數：
        // 1、目錄名稱 2、目錄文本信息 
        // 3、文件夾權限，Ids.OPEN_ACL_UNSAFE表示所有權限 
        // 4、目錄類型，CreateMode.EPHEMERAL_SEQUENTIAL表示創建臨時目錄，session斷開連接則目錄自動刪除
        String createdPath = zk.create(
                "/" + clusterNode + "/" + serverNode, 
                address.getBytes("utf-8"), 
                Ids.OPEN_ACL_UNSAFE, 
                CreateMode.EPHEMERAL_SEQUENTIAL);

采用CreateMode.EPHEMERAL_SEQUENTIAL模式創建臨時節點，可以支持服務節點的實時管理。沒錯，這個模式和《Zookeeper使用實例——分布式共享鎖》中創建有序節點支持分布式共享鎖是一致的。EPHEMERAL_SEQUENTIAL表示創建有序的臨時目錄節點，zookeeper客戶端創建臨時節點后，只要session斷開，該臨時節點會自動刪除。

所以，服務器在zookeeper上創建一個臨時目錄節點，通過節點事件監聽我們可以知道服務器已經加入到服務網絡中，監聽到臨時目錄節點刪除事件，我們可以知道該節點對應的服務器已經脫離服務網絡。下面我們看看具體代碼

1、 服務器啟動后在zookeeper創建臨時目錄

package com.coshaho.learn.zookeeper;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;

/**
 * 
 * 服務節點啟動后注冊到zookeeper
 * @author coshaho
 *
 */
public class AppServer extends Thread
{
    private String clusterNode = "Locks";
    private String serverNode = "mylock";
    private String serverName;
    private long sleepTime;

    public void run()
    {
        try 
        {
            connectZookeeper(serverName);
        } 
        catch (Exception e) 
        {
            e.printStackTrace();
        }
    }
    public void connectZookeeper(String address) throws Exception 
    {
        ZooKeeper zk = new ZooKeeper("192.168.1.104:12181", 5000, new Watcher() 
        {
            public void process(WatchedEvent event) 
            {}
        });

        // 關鍵方法，創建包含自增長id名稱的目錄，這個方法支持了分布式鎖的實現
        // 四個參數：
        // 1、目錄名稱 2、目錄文本信息 
        // 3、文件夾權限，Ids.OPEN_ACL_UNSAFE表示所有權限 
        // 4、目錄類型，CreateMode.EPHEMERAL_SEQUENTIAL表示創建臨時目錄，session斷開連接則目錄自動刪除
        String createdPath = zk.create(
                "/" + clusterNode + "/" + serverNode, 
                address.getBytes("utf-8"), 
                Ids.OPEN_ACL_UNSAFE, 
                CreateMode.EPHEMERAL_SEQUENTIAL);
        System.out.println("create: " + createdPath);
        Thread.sleep(sleepTime);
    }
    
    public AppServer(String serverName, long sleepTime)
    {
        this.serverName = serverName;
        this.sleepTime = sleepTime;
    }
}

2、 節點管理服務監聽zookeeper臨時目錄節點創建刪除事件

package com.coshaho.learn.zookeeper;

import java.util.ArrayList;
import java.util.List;

import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooKeeper;

/**
 * 
 * 客戶端注冊監聽server節點變化
 * @author coshaho
 *
 */
public class AppMaster 
{
    private String clusterNode = "Locks";
    private ZooKeeper zk;
    private volatile List<String> serverList;

    public void connectZookeeper() throws Exception 
    {
        // 注冊全局默認watcher
        zk = new ZooKeeper("192.168.1.104:12181", 5000, new Watcher() 
        {
            public void process(WatchedEvent event) 
            {
                if (event.getType() == EventType.NodeChildrenChanged 
                        && ("/" + clusterNode).equals(event.getPath())) 
                {
                    try 
                    {
                        updateServerList();
                    } 
                    catch (Exception e) 
                    {
                        e.printStackTrace();
                    }
                }
            }
        });

        updateServerList();
    }

    private void updateServerList() throws Exception 
    {
        List<String> newServerList = new ArrayList<String>();

        // watcher注冊后，只能監聽事件一次，參數true表示繼續使用默認watcher監聽事件
        List<String> subList = zk.getChildren("/" + clusterNode, true);
        for (String subNode : subList)
        {
            // 獲取節點數據
            byte[] data = zk.getData("/" + clusterNode + "/" + subNode, false, null);
            newServerList.add(new String(data, "utf-8"));
        }

        serverList = newServerList;
        System.out.println("server list updated: " + serverList);
    }

    public static void main(String[] args) throws Exception 
    {
        AppMaster ac = new AppMaster();
        ac.connectZookeeper();
        Thread.sleep(Long.MAX_VALUE);
    }
}

3、 啟動兩個服務器

package com.coshaho.learn.zookeeper;

public class Server1 
{
    public static void main(String[] args) throws Exception 
    {
        AppServer server1 = new AppServer("Server1", 5000);
        server1.start();
    }
}

package com.coshaho.learn.zookeeper;

public class Server2 
{
    public static void main(String[] args) throws Exception 
    {
        AppServer server1 = new AppServer("Server2", 10000);
        server1.start();
    }
}

4、 運行結果

log4j:WARN No appenders could be found for logger (org.apache.zookeeper.ZooKeeper).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
server list updated: []
server list updated: [Server1]
server list updated: [Server2, Server1]
server list updated: [Server2]
server list updated: []