轻量级RPC分布式网络通信框架设计——服务配置中心zookeeper解析

分布式网络通信框架设计——服务配置中心zookeeper解析"/>

轻量级RPC分布式网络通信框架设计——服务配置中心zookeeper解析

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   /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);}
}

查询服务

• 通过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();   // 启动zkservervoid 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

#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  pollwatcher回调线程 pthread_create*/// 异步操作 知道watcher结束才成功m_zhandle = zookeeper_init(connstr.c_str(), global_watcher, 30000, nullptr, nullptr, 0);  //连接字符串 全局回调  会话超时时间30sif (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)]

注册中心的高可用

• 如果注册中心只有一台机器，一旦挂了，整个系统就宕了。所以需要多台机器来保证高可用性。这样引出了新的问题，比如树形结构需要在多台机器之间进行同步，通信超时了怎么办，如何保证树形结构在机器之间的强一致性。