本文介紹下zookeeper中leader選舉機制的基本用法和關鍵知識點。

一、 選項設定
提到Leader選舉，先需要重點介紹下建立znode時的Flag選項。

• ZOO_EPHEMERAL
  

ZOO_EPHEMERAL，用來標記當建立這個znode的節點和Zookeeper失去連線後，這個znode將不再存在在Zookeeper裡，Zookeeper使用Watcher察覺事件資訊。當客戶端接收到事件資訊，比如連線超時、節點資料改變、子節點改變，可以呼叫相應的行為來處理資料。

• ZOO_SEQUENCE

ZOO_SEQUENCE 用來標識節點命名具有遞增的字尾序號(一般是節點名稱後填充 10 位字元的序號，如 /xyz0000000000, /xyz0000000001, /xyz0000000002, ...)，如下便所示，直接在/app_watch/下建立節點。

[zk: localhost:2181(CONNECTED) 42] ls /app_watch
[0000000017, 0000000018]

同樣地，ZOO_EPHEMERAL, ZOO_SEQUENCE 可以組合使用，下面的示例中就用到了序列號的特性。

二、監視機制

以下面的示例中，可以看到在client的回撥中，每監視到一次事件通知，需要再次呼叫觸發監視，這裡就需要具體說明下zookeeper和 watch機制。

Zookeeper 中最有特色且最不容易理解的是監視(Watches)。Zookeeper 所有的讀操作——getData(), getChildren(), 和 exists() 都 可以設定監視(watch)，監視事件可以理解為一次性的觸發器， 官方定義如下： a watch event is one-time trigger, sent to the client that set the watch, which occurs when the data for which the watch was set changes。對此需要作出如下理解：

• （一次性觸發）One-time trigger

當設定監視的資料發生改變時，該監視事件會被髮送到客戶端，例如，如果客戶端呼叫了 getData("/znode1", true) 並且稍後 /znode1 節點上的資料發生了改變或者被刪除了，客戶端將會獲取到 /znode1 發生變化的監視事件，而如果 /znode1 再一次發生了變化，除非客戶端再次對 /znode1 設定監視，否則客戶端不會收到事件通知。

• （傳送至客戶端）Sent to the client

Zookeeper 客戶端和服務端是通過 socket 進行通訊的，由於網路存在故障，所以監視事件很有可能不會成功地到達客戶端，監視事件是非同步傳送至監視者的，Zookeeper 本身提供了保序性(ordering guarantee)：即客戶端只有首先看到了監視事件後，才會感知到它所設定監視的 znode 發生了變化(a client will never see a change for which it has set a watch until it first sees the watch event). 網路延遲或者其他因素可能導致不同的客戶端在不同的時刻感知某一監視事件，但是不同的客戶端所看到的一切具有一致的順序。

• （被設定 watch 的資料）The data for which the watch was set

 這意味著 znode 節點本身具有不同的改變方式。你也可以想象 Zookeeper 維護了兩條監視連結串列：資料監視和子節點監視(data watches and child watches) getData() and exists() 設定資料監視，getChildren() 設定子節點監視。 或者，你也可以想象 Zookeeper 設定的不同監視返回不同的資料，getData() 和 exists() 返回 znode 節點的相關資訊，而 getChildren() 返回子節點列表。因此， setData() 會觸發設定在某一節點上所設定的資料監視(假定資料設定成功)，而一次成功的 create() 操作則會出發當前節點上所設定的資料監視以及父節點的子節點監視。一次成功的 delete() 操作將會觸發當前節點的資料監視和子節點監視事件，同時也會觸發該節點父節點的child watch。

Zookeeper 中的監視是輕量級的，因此容易設定、維護和分發。當客戶端與 Zookeeper 伺服器端失去聯絡時，客戶端並不會收到監視事件的通知，只有當客戶端重新連線後，若在必要的情況下，以前註冊的監視會重新被註冊並觸發，對於開發人員來說 這通常是透明的。只有一種情況會導致監視事件的丟失，即：通過 exists() 設定了某個 znode 節點的監視，但是如果某個客戶端在此 znode 節點被建立和刪除的時間間隔內與 zookeeper 伺服器失去了聯絡，該客戶端即使稍後重新連線 zookeeper伺服器後也得不到事件通知。

三、示例

在這個示例中，伺服器啟動會在/app_watch下注冊自己的資訊，並使用自增序列號的機制，客戶端同樣監聽/app_watch下的節點資訊變化，並打印出來，如果需要選舉leader的話，直接把child裡的序列號最小的選為leader就好了。

客戶端程式碼如下：

#include <zookeeper.h>
#include <zookeeper_log.h>
#include <iostream>
#include <string.h>

using namespace std;

const char* host = "127.0.0.1:2181";
const int timeout = 2000;
zhandle_t* zkhandle = NULL;
bool is_connect = false;

    char path[] = "/app_watch";
    int ret = 0;
	char buffer[1024];
	int buff_len;
	Stat stat;

void clients_watcher_g(zhandle_t * zh, int type, int state, const char* path, void* watcherCtx) {
    printf("global watcher - type:%d,state:%d\n", type, state);
    if ( type==ZOO_SESSION_EVENT ) {
        if ( state==ZOO_CONNECTED_STATE ) {
            printf("connected to zookeeper service successfully!\n");
            printf("timeout:%d\n", zoo_recv_timeout(zh));
            is_connect = true;
        }
        else if ( state==ZOO_EXPIRED_SESSION_STATE ) {
            printf("zookeeper session expired!\n");
        }
    }
    else {
        printf("other type:%d\n", type);
    }
}

void child_watch_cb(zhandle_t* zh, int type, int state, const char* path, void* watcher) {
    printf("call child watch cb\n");
}

void watch_cb(zhandle_t* zh, int type, int state, const char* path, void* watcher) {
    printf("call watch cb\n");

    // watch path
	buff_len = sizeof(buffer);
	bzero(buffer, buff_len);
    ret = zoo_wget(zkhandle, path, watch_cb, NULL, buffer, &buff_len, &stat);
	printf("ret=%d stat=%d len=%d buffer=%s\n", ret, stat.ctime, buff_len, buffer);
}

int main(int argc, char* argv[]) 
{
    //Init and connect
    zoo_set_debug_level(ZOO_LOG_LEVEL_WARN);
    if (zkhandle) 
        zookeeper_close(zkhandle);

    zkhandle = zookeeper_init(host, clients_watcher_g, timeout, 0, NULL, 0);
    if (NULL == zkhandle) {
        printf("zookeeper init error\n");
        return 0;
    }

    // watch path
	buff_len = sizeof(buffer);
	bzero(buffer, buff_len);
    ret = zoo_wget(zkhandle, path, watch_cb, NULL, buffer, &buff_len, &stat);
	printf("ret=%d stat=%d len=%d buffer=%s\n", ret, stat.ctime, buff_len, buffer);

    while(1) {
    struct String_vector str_vec;
    ret = zoo_wget_children(zkhandle, path, child_watch_cb, NULL, &str_vec);
    if (ZOK != ret) {
        printf("zookeeper wget error\n");
        return 0;
    }
    for (int i = 0; i < str_vec.count; i++) {
        printf("data[%d] data:[%s]\n", i, str_vec.data[i]);
    }
    printf("sleep...\n");
    sleep(5);
    }

    zookeeper_close(zkhandle);
}

#include <zookeeper.h>
#include <zookeeper_log.h>
#include <iostream>
#include <string.h>

using namespace std;

const char* host = "127.0.0.1:2181";
const int timeout = 2000;
zhandle_t* zkhandle = NULL;
bool is_connect = false;

void clients_watcher_g(zhandle_t * zh, int type, int state, const char* path, void* watcherCtx) {
	printf("global watcher - type:%d,state:%d\n", type, state);
	if ( type==ZOO_SESSION_EVENT ) {
		if ( state==ZOO_CONNECTED_STATE ) {
			printf("connected to zookeeper service successfully!\n");
			printf("timeout:%d\n", zoo_recv_timeout(zh));
			is_connect = true;
		}
		else if ( state==ZOO_EXPIRED_SESSION_STATE ) {
			printf("zookeeper session expired!\n");
		}
	}
	else {
		printf("other type:%d\n", type);
	}
}

int main(int argc, char* argv[]) 
{
	//Init and connect
	zoo_set_debug_level(ZOO_LOG_LEVEL_WARN);
	if (zkhandle) 
		zookeeper_close(zkhandle);

	zkhandle = zookeeper_init(host, clients_watcher_g, timeout, 0, NULL, 0);
	if (NULL == zkhandle) {
		printf("zookeeper init error\n");
		return 0;
	}

    // init
    char path[] = "/app_watch";
	Stat stat;
	int ret = 0;

    // create
    ret = zoo_create(zkhandle, path, NULL, -1, &ZOO_OPEN_ACL_UNSAFE, ZOO_EPHEMERAL|ZOO_SEQUENCE, NULL, 0); 
    printf("create [%s] ret=%d\n", path, ret);

	char buffer[1024];
	int buff_len;
    char child_path[1024];
    //snprintf(child_path, 1023, "%s", path, getpid());
    snprintf(child_path, 1023, "%s/", path, getpid());

    // create
	buff_len = sizeof(buffer);
    snprintf(buffer, buff_len, "10.101.1.101:%d", getpid());
    buff_len = strlen(buffer);
    printf("buff:%s len:%d\n", buffer, buff_len);
    ret = zoo_create(zkhandle, child_path, buffer, buff_len, &ZOO_OPEN_ACL_UNSAFE, ZOO_EPHEMERAL|ZOO_SEQUENCE, NULL, 0); 
    printf("create [%s] ret=%d\n", child_path, ret);

    while(1) sleep(1200);

	zookeeper_close(zkhandle);
}

[zk: localhost:2181(CONNECTED) 11] ls /app_watch            
[0000000026]
[zk: localhost:2181(CONNECTED) 12] get /app_watch/0000000026
10.101.1.101:20594

看到 app_watch下已經註冊了一個節點，id是26，檢視client的日誌如下：

[[email protected] /home/derrywang/zookeeper/zk_demo2]# ./watch_client 
global watcher - type:-1,state:3
ret=0 stat=-1597853671 len=18 buffer=10.101.1.101:29413
connected to zookeeper service successfully!
timeout:4000
data[0] data:[0000000026]
sleep...

再啟動一個伺服器例項，檢視如下：

[zk: localhost:2181(CONNECTED) 13] ls /app_watch            
[0000000026, 0000000027]
[zk: localhost:2181(CONNECTED) 14] get /app_watch/0000000027
10.101.1.101:20871

看到app_watch下已經註冊了兩個節點了，id是27， 檢視client的日誌如下：

call child watch cb
data[0] data:[0000000026]
data[1] data:[0000000027]
sleep...

已經觸發了回撥，並獲了一個一個新的節點，這裡把第一個例項停掉，過一段時間client日誌如下：

call child watch cb
data[0] data:[0000000027]
sleep...

上面的例子說明了zookeeper leader選舉的核心機制了，瞭解了這個機制，再做路由配置等都是應用層的實現了。