zk系列-c++下zookeeper使用实例

ZooKeeper是一个分布式的，开放源码的分布式应用程序协调服务。分布式应用可以使用它来实现诸如：统一命名服务、配置管理、分布式锁服务、集群管理等功能。公司常用到的是Java服务集群的管理。

1.函数介绍

[cpp] 
view plain
copy

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

2.实例

用上面3个函数，就能创建一个简单的集群管理。

数据存储结构为

服务端向Zk注册服务

[cpp] 
view plain
copy

1. //连接zk
2. void ZkRegistry::ConnectZK() {
3. if (zhandle_) {
4. zookeeper_close(zhandle_);
5. }
6. int count = 0;
7. do {
8. ++count;
9. zhandle_ = zookeeper_init(zk_hosts_.c_str(),InitWatcher, timeout_, NULL, NULL, 0);
10. } while (!connected_ && count < ZK_MAX_CONNECT_TIMES);
11. if (count >= ZK_MAX_CONNECT_TIMES) {
12. SLOG_WARN("ZkRegistry::Init --> connect host " << zk_hosts_ << " over max times:" << count);
13. return;
14. }
15. }
16. //发布服务，建立临时节点
17. void ZkRegistry::PublishService() {
18. if (zhandle_ == NULL) {
19. ConnectZK();
20. }
21. string server_path = PING_SERVER + "/" + PingConfig::instance().get_index() + "/"
22. + GetIp() + ":" + PingConfig::instance().get_port();
23. char res_path[128];
24. int rc = zoo_create(zhandle_, server_path.c_str(), GetIp().c_str(), GetIp().size(),
25. &ZOO_OPEN_ACL_UNSAFE, ZOO_EPHEMERAL, res_path, 128);
26. if (rc) {
27. SLOG_INFO("ZkRegistry::PublishService --> zoo_create() path=" << server_path << "," << zerror(rc));
28. }
29. }
30. //每隔10s注册一次服务
31. void ZkRegistry::run() {
32. while(true) {
33. SLOG_INFO("publish service");
34. ConnectZK();
35. PublishService();
36. sleep(10);
37. }
38. }
39. void ZkRegistry::InitWatcher(zhandle_t *zh, int type, int state, const char *path, void *watcher_ctx) {
40. if (state == ZOO_CONNECTED_STATE) {
41. connected_ = true;
42. SLOG_INFO("InitWatcher() ZOO_CONNECTED_STATE");
43. } else if (state == ZOO_AUTH_FAILED_STATE) {
44. SLOG_INFO("InitWatcher() ZOO_AUTH_FAILED_STATE");
45. } else if (state == ZOO_EXPIRED_SESSION_STATE) {
46. SLOG_INFO("InitWatcher() ZOO_EXPIRED_SESSION_STATE");
47. } else if (state == ZOO_CONNECTING_STATE) {
48. SLOG_INFO("InitWatcher() ZOO_CONNECTING_STATE");
49. } else if (state == ZOO_ASSOCIATING_STATE) {
50. SLOG_INFO("InitWatcher() ZOO_ASSOCIATING_STATE");
51. }
52. }

客户端获取服务列表

[cpp] 
view plain
copy

1. //连接zk server
2. void ZkClient::ConnectZK() {
3. cout << "ZkClient::ConnectZK" << endl;
4. if (zhandle_) {
5. zookeeper_close(zhandle_);
6. }
7. zhandle_ = NULL;
8. connected_ = false;
9. int count = 0;
10. do {
11. ++count;
12. zhandle_ = zookeeper_init(zk_hosts_.c_str(), InitWatcher, timeout_, NULL, NULL, 0);
13. sleep(5 * ONE_SECONDS);
14. } while (!connected_ && count < ZK_MAX_CONNECT_TIMES);
15. if (count >= ZK_MAX_CONNECT_TIMES){
16. cout << "ZkClient::Init --> connecting zookeeper host: " << zk_hosts_ << " over times: " << count << endl;
17. }
18. }
19. //更新服务列表，冷备和热备
20. void ZkClient::Update() {
21. cout << "ZkClient::Update" << endl;
22. if (zhandle_ == NULL || connected_ == false) {
23. Init();
24. }
25. //获得服务份数
26. struct String_vector str_vec;
27. int ret = zoo_wget_children(zhandle_, PING_SERVER.c_str(), ServiceWatcher, NULL, &str_vec);
28. if (ret) {
29. cout << "Update --> read path:" << PING_SERVER << " wrong, " << zerror(ret) << endl;
30. return;
31. }
32. //获得各份服务ip:port
33. for (int i = 0; i < str_vec.count; ++i) {
34. struct String_vector node_vec;
35. string path = PING_SERVER + "/" + str_vec.data[i];
36. int ret = zoo_wget_children(zhandle_, path.c_str(), ServiceWatcher, NULL, &node_vec);
37. cout << "Update --> path:" << path << ", ret:" << ret << ", node‘s size:" << node_vec.count << endl;
38. if (ret || node_vec.count != 1) {
39. continue;
40. }
41. ....
42. }
43. }
44. //监控服务变化
45. void ZkClient::ServiceWatcher(zhandle_t *zh, int type, int state, const char *path, void *watcherCtx) {
46. cout << "type:" << type << endl;
47. cout << "state:" << state << endl;
48. cout << "path:" << path << endl;
49. //  cout << "watcherCtx:" << (char*)watcherCtx << endl;
50. cout << "ZOO_CHILD_EVENT:" << ZOO_CHILD_EVENT << endl;
51. if (ZOO_CHILD_EVENT == type) {
52. cout << "ServiceWatcher ZOO_CHILD_EVENT" << endl;
53. ZkClient::Instance().Update();//更新服务列表
54. }
55. }

服务端会每隔一段时间重新注册自己；

客户端在第一次与zk建立连接获取服务列表时，注册监听函数。zk当节点发生变化时，通知客户端，客户端重新获取服务列表，并注册事件。