苏州网站建设设计公司,摄影网站建设任务书,wordpress设计的网站,室内设计平面图手绘Zookeeper概念及其作用
ZooKeeper是一个分布式的#xff0c;开放源码的分布式应用程序协调服务#xff0c;是Google的Chubby一个开源的实现#xff0c;是大数据生态中的重要组件。它是集群的管理者#xff0c;监视着集群中各个节点的状态根据节点提交的反馈进行下一步合理…Zookeeper概念及其作用
ZooKeeper是一个分布式的开放源码的分布式应用程序协调服务是Google的Chubby一个开源的实现是大数据生态中的重要组件。它是集群的管理者监视着集群中各个节点的状态根据节点提交的反馈进行下一步合理操作。最终将简单易用的接口和性能高效、功能稳定的系统提供给用户。
它是一个为分布式应用提供一致性协调服务的中间件 zookeeper入门参考链接https://www.cnblogs.com/xinyonghu/p/11031729.html
在分布式系统中zookeeper提供了非常丰富的应用本文只是剖析其中一小部分但也是非常重要的一个部分即服务注册和服务发现。 在RPC框架中如果没有服务注册和服务发现那么这个RPC框架几乎变得不实用浅显的思路是在RPCConsumer(服务调用端维护一个服务的列表这个列表包含了所有分布式节点服务的ip和端口但考虑这么一种情况如果其中某个节点由于某种原因down掉了或者将这个节点的服务删除了但是RPCConsumer本地还维护的列表中还存在这个服务结点并且还尝试请求这个服务那么显然会调用出错。
类似这种肯定需要动态的维护每个分布式服务节点的状态在该节点down掉或者被撤销时应及时删除这个服务避免RPC调用端继续请求不存在的服务。这就是zookeeper服务注册和服务发现所做的事。 Zookeeper组织数据的格式类似于一个文件系统每个znode结点都可以是一个分布式服务结点一般组织的结构是XXXXService/login、 XXXXService/registe即service_name/method_nameznode结点的数据就是该服务所在节点的ip和port
Zookeeper服务注册和发现的流程
step1Rpc服务端先通过zkClient向zkServer端注册服务也即创建XXXXService/login、 XXXXService/registe节点并填充相应的数据。 step2Rpc调用端再调用某个服务之前通过zkClient向ZkServer查询这个服务节点是否存在如果存在则返回这个服务节点的ip和port。然后进行远程rpc调用否则返回错误终止调用过程。 step3这一步其实zookeeper已经帮我们做了step1中注册服务的过程中zkServer会与这个节点建立一个session并且zkServer以1/3 * timeout 的时间定期为每个与之简历的节点发送心跳包如果得不到回应那么zkServer会认为这个节点已经不存在了会动态的把这个节点上的所有服务都进行删除。
RPC框架引入zookeeper
1、封装zkclient用于与zkServer通信的句柄、例如创建结点和删除结点、以及一些心跳回调操作
#pragma once#include semaphore.h
#include zookeeper/zookeeper.h
#include stringclass ZkClient
{
public: ZkClient();~ZkClient();// zkClient启动连接zkservervoid Start();// 在zkserver上根据指定的path创建Znode节点void Create(const char *path, const char* data, int datalen, int state0);// 根据参数指定的znode节点路径获取znode节点的值std::string GetData(const char* path);private:// zk客户端句柄zhandle_t *m_zhandle;
};// .cc
#include zookeeperutil.h
#include rpcapplication.h
#include iostream//全局的watcher观察器 zkserver给zkclient的通知回调
void global_watcher(zhandle_t *zh, int type, int state, const char* path, void *watcherCtx)
{if(type ZOO_SESSION_EVENT) //回调的消息类型是和会话相关的消息类型{if(state ZOO_CONNECTED_STATE) //zkserver和zkclient连接成功{sem_t *sem (sem_t*) zoo_get_context(zh);sem_post(sem);}}
}ZkClient::ZkClient():m_zhandle(nullptr)
{}
ZkClient::~ZkClient()
{if(m_zhandle ! nullptr){zookeeper_close(m_zhandle); //关闭句柄 释放资源}
}
// zkClient启动连接zkserver
void ZkClient::Start()
{std::string host RpcApplication::GetInstance().GetConfig().Load(zookeeperip);std::string port RpcApplication::GetInstance().GetConfig().Load(zookeeperport);std::string connstr host : port;/*zookeeper_mt多线程版本zookeeper的API客户端程序提供了三个线程APT调用线程网络I/O线程 pthread_create (使用的poll-IO多路复用watcher回调线程 pthread_create*/m_zhandle zookeeper_init(connstr.c_str(), global_watcher, 30000, nullptr, nullptr, 0);if(nullptr m_zhandle){std::cout zookeeper_init error ! std::endl;exit(EXIT_FAILURE);}sem_t sem;sem_init(sem, 0, 0);zoo_set_context(m_zhandle, sem);sem_wait(sem);std::cout zookeeper_init success ! std::endl;
}// 在zkserver上根据指定的path创建Znode节点
void ZkClient::Create(const char *path, const char* data, int datalen, int state)
{char path_buffer[128];int bufferlen sizeof(path_buffer);int flag;//先判断path表示的znode节点是否存在 如果存在 就不能重复创建了flag zoo_exists(m_zhandle, path, 0, nullptr);if(ZNONODE flag) //表示path的znode节点不存在{// 创建指定path的znode节点flag zoo_create(m_zhandle, path, data, datalen, ZOO_OPEN_ACL_UNSAFE, state, path_buffer, bufferlen);if(flag ZOK){std::cout znode create success .... path: path std::endl;}else{std::cout flag : flag std::endl;std::cout znode create error...path: path std::endl;exit(EXIT_FAILURE);}}
}// 根据参数指定的znode节点路径获取znode节点的值
std::string ZkClient::GetData(const char* path)
{char buffer[64];int bufferlen sizeof(buffer);int flag zoo_get(m_zhandle, path, 0, buffer, bufferlen, nullptr);if(flag ! ZOK){std::cout get znode error ...... path path std::endl;return ;}else{return buffer;}
}2、在RPCProvider端进行服务注册
//把当前rpc节点上要发布的服务全部注册到zk上面 让rpc client可以从zk上发现服务// session timeout 30s zkclient 的网络I/O线程 会定时以1/3 * timeout 时间去给zkserver发送ping心跳包ZkClient zkCli;zkCli.Start();//service name为永久性节点 method name 为临时性节点for(auto sp : m_serviceMap){// /service_name --- /UserServiceRPcstd::string service_path / sp.first;zkCli.Create(service_path.c_str(), nullptr, 0);for(auto mp : sp.second.m_methodMap){// /service_name/method_name /UserServiceRPc/Login 存储当前这个rpc服务节点主机的ip和portstd::string method_path service_path / mp.first;char method_path_data[128] {0};sprintf(method_path_data, %s:%d, ip.c_str(), port);//ZOO_EPHEMERAL 表示znode是一个临时性节点zkCli.Create(method_path.c_str(), method_path_data, strlen(method_path_data), ZOO_EPHEMERAL);}}3、RPCConsumer端进行服务发现
//rpc调用方想调用service_name的method_name的服务 需要查询zk上该服务所在的host信息ZkClient zkCli;zkCli.Start();// /UserServiceRpc/Loginstd::string method_path / service_name / method_name;// 127.0.0.1:8000std::string host_data zkCli.GetData(method_path.c_str());if(host_data ){controller-SetFailed(method_path is not exist!);return;}int idx host_data.find(:);if(idx -1){controller-SetFailed(method_path address is invalid);return;}std::string ip host_data.substr(0, idx);uint16_t port atoi(host_data.substr(idx 1, host_data.size() - idx).c_str());至此基本上完整RPC应该具备的核心东西都有了。
