概述
01 zookeeper介绍
zookeeper是为分布式应用提供一致性协调服务的中间件
zeekeeper的文件系统
zookeeper提供了一个多层级的节点命名空间（节点称为znode）。与文件系统不同的是，这些znode节点都可以设置关联的数据，而文件系统中只有文件节点可以存放数据而目录不行。zookeeper为了保证高吞吐和低延迟，在内存中维护了这个树状的目录结构，这种特性使得zookeeper不能存放大量的数据，每个节点的存放数据上限为1M。
zookeeper的通知机制
客户端会对某个znode建立一个watcher事件，当这个znode发生变化时，这些客户端会收到zk的通知，然后客户端可以根据znode变化做出业务上的改变
zookeeper的作用
为了支持高并发，OrderService被部署了4份，每个客户端都保存了一份服务提供者的列表，但这个列表是静态的（在配置文件中写死的），如果服务的提供者发生了变化，例如有些机器down了，或者又新增了OrderService的实例，客户端根本不知道，想要得到最新的服务提供者的URL列表，必须手工更新配置文件，很不方便。
解决方案
解除耦合，增加一个中间层 – 注册中心它保存了能提供的服务的名称，以及URL。首先这些服务会在注册中心进行注册，当客户端来查询的时候，只需要给出名称，注册中心就会给出一个URL。所有的客户端在访问服务前，都需要向这个注册中心进行询问，以获得最新的地址。
注册中心可以是树形结构，每个服务下面有若干节点，每个节点表示服务的实例。
注册中心和各个服务实例直接建立Session，要求实例们定期发送心跳，一旦特定时间收不到心跳，则认为实例挂了，删除该实例。
总结
大概有如下功能：
- master节点选举, 主节点down掉后, 从节点就会接手工作, 并且保证这个节点是唯一的,这也就是所谓首脑模式,从而保证我们集群是高可用的
- 统一配置文件管理, 即只需要部署一台服务器, 则可以把相同的配置文件同步更新到其他所有服务器, 此操作在云计算中用的特别多(例如修改了redis统一配置)
- 数据发布与订阅, 类似消息队列MQ
- 分布式锁,分布式环境中不同进程之间争夺资源,类似于多进程中的锁
- 集群管理, 保证集群中数据的强一致性
02 服务配置中心用法
每个rpc服务器端都会向zookeeper服务注册配置中心 传入 ip + port + 服务名字 客户端进程查询获得ip+port
zookeeper介绍
03 源码解析
注册服务
- 通过start函数连接zkserver
- 创建zk路径
// 把当前rpc节点上要发布的服务全部注册到zk上面，让rpc client可以从zk上发现服务
ZkClient zkCli;
zkCli.Start();
// service_name为永久性节点 method_name为临时性节点
for (auto &sp : m_serviceMap)
{
// /service_name /UserServiceRpc
std::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和port
std::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);
}
}
查询服务
- 通过start连接zkserver
- 通过getdata方法传入method_name 获得ip + port
// rpc调用方想调用service_name的method_name服务，需要查询zk上该服务所在的host信息
ZkClient zkCli;
zkCli.Start(); // 连接服务器
// /UserServiceRpc/Login
std::string method_path = "/" + service_name + "/" + method_name;
// 127.0.0.1:8000
std::string host_data = zkCli.GetData(method_path.c_str());
zk代码
一级路径 /service_name 设置为永久性节点
二级路径 /service_name/method_name 设置为临时性节点
zookeeper.util
#pragma once
#include <semaphore.h>
#include <zookeeper/zookeeper.h>
#include <string>
//封装的zk类
class ZkClient {
public:
ZkClient();
~ZkClient();
void Start(); // 启动zkserver
void Create(const char *path, const char *data, int datalen, int state = 0); // 在zkserver上根据指定的path创建znode节点 state = 0代表永久性节点
std::string GetData(const char *path); //根据参数指定的znode节点路径 获得znode节点的值
private:
zhandle_t *m_zhandle; //zk的客户端句柄 操作zkserver
};
zookeeperutil.cc
#include "zookeeperutil.h"
#include "mprpcapplication.h"
#include <semaphore.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) // zkclient和zkserver连接成功
{
sem_t *sem = (sem_t*)zoo_get_context(zh); // 获取信号量
sem_post(sem); // 信号量资源+1
}
}
}
ZkClient::ZkClient() : m_zhandle(nullptr)
{
}
ZkClient::~ZkClient()
{
if (m_zhandle != nullptr)
{
zookeeper_close(m_zhandle); // 关闭句柄，释放资源 MySQL_Conn
}
}
// 连接zkserver
void ZkClient::Start()
{
std::string host = MprpcApplication::GetInstance().GetConfig().Load("zookeeperip");
std::string port = MprpcApplication::GetInstance().GetConfig().Load("zookeeperport");
std::string connstr = host + ":" + port;
/*
zookeeper_mt：多线程版本
zookeeper的API客户端程序提供了三个线程
API调用线程
网络I/O线程 pthread_create poll
watcher回调线程 pthread_create
*/
// 异步操作 知道watcher结束才成功
m_zhandle = zookeeper_init(connstr.c_str(), global_watcher, 30000, nullptr, nullptr, 0); //连接字符串 全局回调 会话超时时间30s
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); // 初始信号量为0 在这阻塞 去global_watcher中获取资源 信号量加1 获取资源成功 进行下一步
std::cout << "zookeeper_init success!" << std::endl; // 异步的连接过程
}
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);
}
}
}
// 根据指定的path，获取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;
}
}
04 其他功能介绍
分布式锁
分布式锁, 多台机器上运行的不同的系统操作同一资源
- 使用Master选举的方式，让大家去抢，谁能抢到就创建一个
/distribute_lock节点，读完以后就删除，让大家再来抢。缺点是某个系统可能多次抢到，不够公平。 - 让每个系统在注册中心的
/distribute_lock下创建子节点，然后编号,每个系统检查自己的编号，谁的编号小认为谁持有了锁，比如下图中是系统1持有了锁
系统1操作完成以后，就可以把process_01删除了，再创建一个新的节点 process_04。此时是process_02最小了，所以认为系统2持有了锁。
操作完成以后也要把process_02节点删除，创建新的节点。这时候process_03就是最小的了，可以持有锁了。
注册中心的高可用
08153)]
系统1操作完成以后，就可以把process_01删除了，再创建一个新的节点 process_04。此时是process_02最小了，所以认为系统2持有了锁。
[外链图片转存中…(img-hrJhiwfU-1667742208153)]
操作完成以后也要把process_02节点删除，创建新的节点。这时候process_03就是最小的了，可以持有锁了。
[外链图片转存中…(img-FANo7o5M-1667742208153)]
注册中心的高可用
- 如果注册中心只有一台机器，一旦挂了，整个系统就宕了。所以需要多台机器来保证高可用性。这样引出了新的问题，比如树形结构需要在多台机器之间进行同步，通信超时了怎么办，如何保证树形结构在机器之间的强一致性。
最后
以上就是淡淡火为你收集整理的轻量级RPC分布式网络通信框架设计——服务配置中心zookeeper解析01 zookeeper介绍02 服务配置中心用法03 源码解析04 其他功能介绍的全部内容，希望文章能够帮你解决轻量级RPC分布式网络通信框架设计——服务配置中心zookeeper解析01 zookeeper介绍02 服务配置中心用法03 源码解析04 其他功能介绍所遇到的程序开发问题。
如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。
发表评论 取消回复