Zookeeper C API学习总结

typedef void (*watcher_fn)(zhandle_t *zh, int type, int state, const char *path,void *watcherCtx);

// 处理返回 void 类型的回调函数
typedef void(* void_completion_t)(int rc, const void *data);

// 处理返回 Stat 结构的回调函数
typedef void(* stat_completion_t)(int rc, const struct Stat *stat, const void *data); 

// 处理返回字符串的回调函数
typedef void(* string_completion_t)(int rc, const char *value, const void *data); 

// 处理返回数据的回调函数
typedef void(* data_completion_t)(int rc, const char *value, int value_len, const struct Stat *stat, const void *data); 

// 处理返回字符串列表(a list of string)的回调函数
typedef void(* strings_completion_t)(int rc, const struct String_vector *strings, const void *data); 

// 同时处理返回字符串列表(a list of string)和 Stat 结构的回调函数
typedef void(* strings_stat_completion_t)(int rc, const struct String_vector *strings, const struct Stat *stat, const void *data); 

// 处理以及返回 ACL 信息的回调函数
typedef void(* acl_completion_t)(int rc, struct ACL_vector *acl, struct Stat *stat, const void *data);

ZOOAPI zhandle_t *zookeeper_init(const char *host, watcher_fn fn,
                                 int recv_timeout,
                                 const clientid_t * clientid,
                                 void *context, int flags);

ZOOAPI void zoo_set_debug_level(ZooLogLevel logLevel);

ZOOAPI void zoo_set_log_stream(FILE * logStream);

ZOOAPI struct sockaddr *zookeeper_get_connected_host(zhandle_t * zh, struct sockaddr
                                                     *addr,
                                                     socklen_t * addr_len);

ZOOAPI int zookeeper_interest(zhandle_t * zh, int *fd, int *interest,
                              struct timeval *tv);

ZOOAPI 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);

ZOOAPI int zoo_delete(zhandle_t * zh, const char *path, int version);

ZOOAPI int zoo_exists(zhandle_t * zh, const char *path, int watch,
                      struct Stat *stat);

ZOOAPI int zoo_get(zhandle_t * zh, const char *path, int watch,
                   char *buffer, int *buffer_len, struct Stat *stat);

ZOOAPI int zoo_get_children(zhandle_t * zh, const char *path,
                            int watch, struct String_vector *strings);

ZOOAPI int zoo_acreate(zhandle_t * zh, const char *path,
                       const char *value, int valuelen,
                       const struct ACL_vector *acl, int flags,
                       string_completion_t completion, const void *data);

ZOOAPI int zoo_awget(zhandle_t * zh, const char *path,
                     watcher_fn watcher, void *watcherCtx,
                     data_completion_t completion, const void *data);

ZOOAPI int zoo_aset(zhandle_t * zh, const char *path,
                    const char *buffer, int buflen, int version,
                    stat_completion_t completion, const void *data);

ZOOAPI int zoo_aget_children(zhandle_t * zh, const char *path,
                             int watch,
                             strings_completion_t completion,
                             const void *data);

ZOOAPI int zoo_awget_children(zhandle_t * zh, const char *path,
                              watcher_fn watcher, void *watcherCtx,
                              strings_completion_t completion,
                              const void *data);

ZOOAPI int zoo_aget_children2(zhandle_t * zh, const char *path,
                              int watch,
                              strings_stat_completion_t completion,
                              const void *data);

ZOOAPI int zoo_awget_children2(zhandle_t * zh, const char *path,
                               watcher_fn watcher, void *watcherCtx,
                               strings_stat_completion_t completion,
                               const void *data);

ZOOAPI int zoo_async(zhandle_t * zh, const char *path,
                     string_completion_t completion, const void *data);

int main(int argc, const char *argv[]) {
  /* 设置zookeeper host，其中host字符串格式为逗号隔开的IP:PORT对 */
  static const char* host = "x.x.x.x:x";
  /* 设置客户端连接服务器超时时间，单位ms */
  int timeout = 30000;
  /* 设置日志为调试级别 */
  zoo_set_debug_level(ZOO_LOG_LEVEL_DEBUG);
  /* 用于启用或停用quarum端点的随机化排序，通常仅在测试时使用。
      如果非0，使得client连接到quarum端按照被初始化的顺序。
      如果是0，zookeeper_init将变更端点顺序，使得client连接分布在更优的端点上。*/
  zoo_deterministic_conn_order(1);
  /* 初始化zookeeper句柄，传入全局监视器回调函数、超时、以及上下文信息 */
  zhandle_t* zkhandle = zookeeper_init(host, zk_watcher_g, 30000, 0, (void *)"hello world", 0);
  if (zkhandle == NULL) {
    fprintf(stderr, "Error when connecting to zookeeper servers...\n");
    exit(EXIT_FAILURE);
  }
  /* 等待异步回调 */
  int fd, interest, events;
  int rc;
  fd_set rfds, wfds, efds;
  FD_ZERO(&rfds);
  FD_ZERO(&wfds);
  FD_ZERO(&efds);
  while (1) {
    struct timeval tv;
    /* 注册zookeeper中的事件 */
    zookeeper_interest(zkhandle, &fd, &interest, &tv);
    if (fd != -1) {
      if (interest & ZOOKEEPER_READ) {
        FD_SET(fd, &rfds);
      } else {
        FD_CLR(fd, &rfds);
      }

      if (interest & ZOOKEEPER_WRITE) {
        FD_SET(fd, &wfds);
      } else {
        FD_CLR(fd, &wfds);
      }
    } else {
      fd = 0;
    }

    if (select(fd + 1, &rfds, &wfds, &efds, &tv) < 0) {
      printf("[%s %d]select failed, err=%d, msg=%s\n", __FUNCTION__, __LINE__,
             errno, strerror(errno));
    }
    events = 0;
    if (FD_ISSET(fd, &rfds)) {
      events |= ZOOKEEPER_READ;
    }
    if (FD_ISSET(fd, &wfds)) {
      events |= ZOOKEEPER_WRITE;
    }
    /* 通知zookeeper客户端事件已经发生 */
    zookeeper_process(zkhandle, events);
  }
  zookeeper_close(zkhandle);
}

void zk_watcher_g(zhandle_t *zh, int type, int state, const char* path, void *watcherCtx) {
  int ret;

  const clientid_t *zk_clientid;
  /* 声明节点路径 */
  if(path == NULL || strlen(path) == 0) {
    path = "/xyz";
  }

  /* 当前event type是会话 */
  if(type == ZOO_SESSION_EVENT) {
    if(state == ZOO_EXPIRED_SESSION_STATE) {
      printf("[%s %d] zookeeper session expired\n", __FUNCTION__, __LINE__);
    }
    /* 会话状态是已连接 */
    else if(state == ZOO_CONNECTED_STATE) {
      /* 获取客户端的 session id，只有在客户端的当前连接状态有效时才可以 */
      zk_clientid = zoo_client_id(zh);
      
      /* 调用异步api */
      ret = zoo_aexists(zh, path, TRUE, zk_stat_completion, path);
      
      ret = zoo_aget(zh, path, TRUE, zk_data_completion, "get param");
      
      ret = zoo_aget_children2(zh, path, TRUE, zk_strings_stat_completion, "get children param");
    }
  }
  /* 当前event type是节点创建事件 */
  else if (type == ZOO_CREATED_EVENT) {
    ...
  }
  /* 当前event type是节点删除事件 */
  else if (type == ZOO_DELETED_EVENT) {
    ...
  }
  /* 当前event type是节点数据改变事件 */
  else if (type == ZOO_CHANGED_EVENT) {
    ...
  }
  /* 当前event type是子节点事件 */
  else if (type == ZOO_CHILD_EVENT) {
    ...
  }
}

Something happened.
type: SESSION_EVENT
state: 4261760s
path: 
watcherCtx: hello world
[zk_watcher_g 140] connected to zookeeper server with clientid=96408539334299438
in stat completion rc = 0 data = /xyz stat:
tctime = Sun Nov 20 10:12:08 2016
 | tczxid = 609cd7384
tmtime = Sun Nov 20 10:12:08 2016
 | tmzxid = 609cd7384
version = 0in data completion []: rc = 0
tctime = Sun Nov 20 10:12:08 2016
 | tczxid = 609cd7384
tmtime = Sun Nov 20 10:12:08 2016
 | tmzxid = 609cd7384
version = 0in strings stat completion [get children param]: rc = 0, string count 0
tctime = Sun Nov 20 10:12:08 2016
 | tczxid = 609cd7384
tmtime = Sun Nov 20 10:12:08 2016
 | tmzxid = 609cd7384

version = 0Something happened.
type: CHILD_EVENT
state: 4261760s
path: /xyz
watcherCtx: hello world
path (null) child event
in data completion []: rc = 0
tctime = Sun Nov 20 10:12:08 2016
 | tczxid = 609cd7384
tmtime = Sun Nov 20 10:12:08 2016
 | tmzxid = 609cd7384
version = 0in strings stat completion [get children param]: rc = 0, string count 1
0: abc
tctime = Sun Nov 20 10:12:08 2016
 | tczxid = 609cd7384
tmtime = Sun Nov 20 10:12:08 2016
 | tmzxid = 609cd7384

version = 0Something happened.
type: CHANGED_EVENT
state: 4261760s
path: /xyz
watcherCtx: hello world
path /xyz changed
in strings stat completion [get children param]: rc = 0, string count 1
0: abc
tctime = Sun Nov 20 10:12:08 2016
 | tczxid = 609cd7384
tmtime = Sun Nov 20 10:16:56 2016
 | tmzxid = 609cd7449
version = 1in data completion []: rc = 0
tctime = Sun Nov 20 10:12:08 2016
 | tczxid = 609cd7384
tmtime = Sun Nov 20 10:16:56 2016
 | tmzxid = 609cd7449

int main(int argc, const char *argv[]) {
  if (argc != 2) {
    printf("usage: zookeeper_m_test server_address\n");
    exit(-1);
  }
  const char* host = argv[1];
  int timeout = 30000;
  char buffer[512];
  int *bufferlen;

  zoo_set_debug_level(ZOO_LOG_LEVEL_DEBUG);
  /* 初始化zookeeper句柄，传入全局监视器回调函数、超时、以及简单参数 */
  zhandle_t* zkhandle = zookeeper_init(host, zookeeper_watcher_g, timeout, 0,
                                       NULL, 0);
  if (zkhandle == NULL) {
    fprintf(stderr, "Error when connecting to zookeeper servers...\n");
    exit(EXIT_FAILURE);
  }
  /* 调用api */
  create(zkhandle, "/abc");
  exists(zkhandle, "/abc");
  getACL(zkhandle, "/abc");
  del(zkhandle, "/abc");
  while (1)
    ;
  zookeeper_close(zkhandle);
}

Something happened.
type: -1
state: 3
path: 
watcherCtx: hello zookeeper.
[acreate]: rc = -110
aexists: rc = 0 Stat:
        ctime = Sun Nov 20 00:25:19 2016
        czxid=609ccd4de
        mtime=Sun Nov 20 00:26:04 2016
        mzxid=609ccd4e9
        version=1       aversion=0
        ephemeralOwner = 0
        
...

Something happened.
type: 3
state: 3
path: /xyz
watcherCtx: hello zookeeper.
...
[adelete]: rc = 0