利用Zookeeper實現分布式鎖及服務注冊中心

對於Zookeeper的定義以及原理，網上已經有很多的優秀文章對其進行了詳細的介紹，所以本文不再進行這方面的闡述。

本文主要介紹一些基本的准備工作以及zookeeper.net的使用。

本文源代碼github地址：https://github.com/Mike-Zrw/ZookeeperHelper

zookeeper下載地址：https://archive.apache.org/dist/zookeeper/

ZooInspector下載地址：https://issues.apache.org/jira/secure/attachment/12436620/ZooInspector.zip

Zookeeper下載及安裝

目前版本最新的為3.5.3 。 本文使用的版本為3.4.10

隨意下載一個版本的壓縮包，解壓到某個文件夾中. zookeeper服務的啟動腳本在bin目錄下的文件：zkServer.cmd

在服務啟動之前，需要對配置文件進行基本的設置：

將conf目錄下的zoo_sample.cfg改名為 zoo.cf

修改里面的日志文件路徑，我修改完之后的文件內容如下

The number of milliseconds of each tick

tickTime=2000

The number of ticks that the initial

synchronization phase can take

initLimit=10

The number of ticks that can pass between

sending a request and getting an acknowledgement

syncLimit=5

the directory where the snapshot is stored.

do not use /tmp for storage, /tmp here is just

example sakes.

dataDir=data
dataLogDir=data\log

the port at which the clients will connect

clientPort=2181

the maximum number of client connections.

increase this if you need to handle more clients

maxClientCnxns=60

Be sure to read the maintenance section of the

administrator guide before turning on autopurge.

http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance

The number of snapshots to retain in dataDir

autopurge.snapRetainCount=3

Purge task interval in hours

Set to "0" to disable auto purge feature

autopurge.purgeInterval=1

• tickTime：Zookeeper之間維持心跳的時間間隔
• initLimit：集群中的follower服務器(F)與leader服務器(L)之間 初始連接 時能容忍的最多心跳數（tickTime的數量），這里為2000*10 ,超過10次心跳仍然沒同步完成，會重新選舉leader
• syncLimit：集群中的follower服務器(F)與leader服務器(L)之間 請求和應答 之間能容忍的最多心跳數（tickTime的數量）。超時follwer將被丟棄
• dataDir：Zookeeper保存數據的目錄 ，目錄不可含中文字符
• dataLogDir：Zookeeper保存日志文件的目錄
• clientPort：Zookeeper服務器的端口

配置完成之后就可以雙擊zkServer.cmd啟動zookeeper服務了

ZooInspector的安裝

ZooInspector是zookeeper的監視工具，可以查看zookeeper的數據信息

下載完成后直接解壓，運行zookeeper-dev-ZooInspector.jar。如果默認端口沒有修改，直接點擊連接就可以了

ZooKeeper.Net

通過Nuget來安裝ZooKeeper.Net的開發包到項目中

下面的代碼會建立一個zookeeper的連接，並創建一個名為parent的臨時目錄

   public static void TestConnect()
    {
        Console.WriteLine("建立連接");
        //服務地址為：localhost:2181  超時連接30秒
        using (ZooKeeper Instance = new ZooKeeperNet.ZooKeeper("localhost:2181", new TimeSpan(0, 0, 30), new Watcher("new")))
        {
            Console.WriteLine("檢測是否有parent目錄");
            var sdata = Instance.Exists("/parent", new Watcher("exists"));
            Console.WriteLine(sdata==null?"否":"是");
            if (sdata==null)
            {
                Console.WriteLine("開始創建parent目錄");
                //data：目錄關聯的data為：this is the parentnode data
                //acl：目錄的訪問權限控制
                //CreateMode：Ephemeral：目錄為臨時目錄，斷開連接目錄會自動清除  永久目錄：Persistent  自增目錄：***Sequential
                Instance.Create("/parent", "this is the parentnode data".GetBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.Ephemeral);
                Console.WriteLine("創建parent目錄完成");
                Console.WriteLine("檢測是否有parent目錄");
                sdata = Instance.Exists("/parent", new Watcher("exists2"));
                Console.WriteLine(sdata == null ? "否" : "是");
                if (sdata != null)
                {
                    Console.WriteLine("刪除parent目錄");
                    Instance.Delete("/parent", 0);
                }
            }
        }
    }

運行結果如下

需要注意的是執行這段代碼可能會報錯提示connection loss，出現錯誤的原因是建立連接的這一步是異步的操作，應當等待zookeeper連接成功之后再執行下面的代碼。所以這段代碼只是進行一個簡單的演示，nuget上附有正確的使用方式

關於zookeeper的watch定義如下：

監視是一種簡單的機制，使客戶端收到關於ZooKeeper集合中的更改的通知。客戶端可以在讀取特定znode時設置Watches。Watches會向注冊的客戶端發送任何znode（客戶端注冊表）更改的通知。

Znode更改是與znode相關的數據的修改或znode的子項中的更改。只觸發一次watches。如果客戶端想要再次通知，則必須通過另一個讀取操作來完成。當連接會話過期時，客戶端將與服務器斷開連接，相關的watches也將被刪除。

Zookeeper的每個節點稱之為znode，znode的類型分為四種：持久節點，臨時節點，持久順序節點，臨時順序節點

這里簡單說下順序節點：比如我需要建立一個名稱為 /app的節點，那么zookeeper會建立一個名為/app0000000001的節點，如果再有人要創建一個名為/app的順序節點，新建的節點名稱會變為/app0000000002

也就是說順序節點生成之后的名字是指定的名字加十位序列號，序列號不會重復，即使是兩個操作同時創建。 我們可以利用這一個特性來實現分布式鎖：

分布式鎖的偽代碼如下：

public void lock(){
var parentpath=/lock_
var path = CreateEphemeralSequentialNode(parentpath)
while(true){
var children = 獲取父節點的所有子節點
if(path是children中的最小的){
成功獲取鎖
return;
}else{
監控等待前一個節點刪除
wait();
}
}
}

具體的代碼可以在github上查看