ZooKeeper是一個分散式的，開放原始碼的分散式應用程式協調服務。分散式應用可以使用它來實現諸如：統一命名服務、配置管理、分散式鎖服務、叢集管理等功能。公司常用到的是Java服務叢集的管理。

1.函式介紹

//create a handle to used communicate with zookeeper
zhandle_t *zookeeper_init(const char *host, watcher_fn fn, int recv_timeout, const clientid_t *clientid, void *context, int flags)
//create a node synchronously
int zoo_create(zhandle_t *zh, const char *path, const char *value,int valuelen,
   const struct ACL_vector *acl, int flags,char *path_buffer, int path_buffer_len);
//lists the children of a node synchronously.
int zoo_wget_children(zhandle_t *zh, const char *path, watcher_fn watcher, void* watcherCtx, struct String_vector *strings)
//close the zookeeper handle and free up any resources.
int zookeeper_close(zhandle_t *zh)
 

2.例項

用上面3個函式，就能建立一個簡單的叢集管理。

資料儲存結構為

服務端向Zk註冊服務

//連線zk
void ZkRegistry::ConnectZK() {
  if (zhandle_) {
    zookeeper_close(zhandle_);
  }
  int count = 0;
  do {
    ++count;
    zhandle_ = zookeeper_init(zk_hosts_.c_str(),InitWatcher, timeout_, NULL, NULL, 0);
  } while (!connected_ && count < ZK_MAX_CONNECT_TIMES);
 
  if (count >= ZK_MAX_CONNECT_TIMES) {
    SLOG_WARN("ZkRegistry::Init --> connect host " << zk_hosts_ << " over max times:" << count);
    return;
  }
}
 
//釋出服務，建立臨時節點
void ZkRegistry::PublishService() {
  if (zhandle_ == NULL) {
    ConnectZK();
  }
  string server_path = PING_SERVER + "/" + PingConfig::instance().get_index() + "/"
    + GetIp() + ":" + PingConfig::instance().get_port();
  char res_path[128];
  int rc = zoo_create(zhandle_, server_path.c_str(), GetIp().c_str(), GetIp().size(),
      &ZOO_OPEN_ACL_UNSAFE, ZOO_EPHEMERAL, res_path, 128);
  if (rc) {
    SLOG_INFO("ZkRegistry::PublishService --> zoo_create() path=" << server_path << "," << zerror(rc));
  }
}
 
//每隔10s註冊一次服務
void ZkRegistry::run() {
  while(true) {
    SLOG_INFO("publish service");
    ConnectZK();
    PublishService();
    sleep(10);
  }
}
 
void ZkRegistry::InitWatcher(zhandle_t *zh, int type, int state, const char *path, void *watcher_ctx) {
  if (state == ZOO_CONNECTED_STATE) {
    connected_ = true;
    SLOG_INFO("InitWatcher() ZOO_CONNECTED_STATE");
  } else if (state == ZOO_AUTH_FAILED_STATE) {
    SLOG_INFO("InitWatcher() ZOO_AUTH_FAILED_STATE");
  } else if (state == ZOO_EXPIRED_SESSION_STATE) {
    SLOG_INFO("InitWatcher() ZOO_EXPIRED_SESSION_STATE");
  } else if (state == ZOO_CONNECTING_STATE) {
    SLOG_INFO("InitWatcher() ZOO_CONNECTING_STATE");
  } else if (state == ZOO_ASSOCIATING_STATE) {
    SLOG_INFO("InitWatcher() ZOO_ASSOCIATING_STATE");
  }
}
 

客戶端獲取服務列表

//連線zk server
void ZkClient::ConnectZK() {
  cout << "ZkClient::ConnectZK" << endl;
  if (zhandle_) {
    zookeeper_close(zhandle_);
  }
  zhandle_ = NULL;
  connected_ = false;
 
  int count = 0;
  do {
    ++count;
    zhandle_ = zookeeper_init(zk_hosts_.c_str(), InitWatcher, timeout_, NULL, NULL, 0);
    sleep(5 * ONE_SECONDS);
  } while (!connected_ && count < ZK_MAX_CONNECT_TIMES);
 
  if (count >= ZK_MAX_CONNECT_TIMES){
    cout << "ZkClient::Init --> connecting zookeeper host: " << zk_hosts_ << " over times: " << count << endl;
  }
}
//更新服務列表，冷備和熱備
void ZkClient::Update() {
  cout << "ZkClient::Update" << endl;
  if (zhandle_ == NULL || connected_ == false) {
    Init();
  }
  //獲得服務份數
  struct String_vector str_vec;
  int ret = zoo_wget_children(zhandle_, PING_SERVER.c_str(), ServiceWatcher, NULL, &str_vec);
  if (ret) {
    cout << "Update --> read path:" << PING_SERVER << " wrong, " << zerror(ret) << endl;
    return;
  }
 
  //獲得各份服務ip:port
  for (int i = 0; i < str_vec.count; ++i) {
    struct String_vector node_vec;
    string path = PING_SERVER + "/" + str_vec.data[i];
    int ret = zoo_wget_children(zhandle_, path.c_str(), ServiceWatcher, NULL, &node_vec);
    cout << "Update --> path:" << path << ", ret:" << ret << ", node's size:" << node_vec.count << endl;
    if (ret || node_vec.count != 1) {
      continue;
    }
    ....
  }
}
//監控服務變化
void ZkClient::ServiceWatcher(zhandle_t *zh, int type, int state, const char *path, void *watcherCtx) {
  cout << "type:" << type << endl;
  cout << "state:" << state << endl;
  cout << "path:" << path << endl;
//  cout << "watcherCtx:" << (char*)watcherCtx << endl;
  cout << "ZOO_CHILD_EVENT:" << ZOO_CHILD_EVENT << endl;
  if (ZOO_CHILD_EVENT == type) {
    cout << "ServiceWatcher ZOO_CHILD_EVENT" << endl;
    ZkClient::Instance().Update();//更新服務列表    
  }
}
 

服務端會每隔一段時間重新註冊自己；

客戶端在第一次與zk建立連接獲取服務列表時，註冊監聽函式。zk當節點發生變化時，通知客戶端，客戶端重新獲取服務列表，並註冊事件。