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

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

查询服务

• 通过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());
1
2
3
4
5
6
7

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
};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105

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)]

注册中心的高可用

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