zookeeper分布式协调服务

• 为啥要zookeeper
• Zookeeper提供了什么
• zookeeper客户端常用命令
• 分布式系统的问题
• • 客户端和服务端提供者的紧耦合
  • 解决
  • 心跳机制(session会话)
  • • 永久性节点
    • 临时性节点（我们所需要的）
• watcher机制(观察者模式)
• 原生ZkClient API存在的问题
• zookeeper的使用

Zookeeper是在分布式环境中应用非常广泛，它的优秀功能很多，比如分布式环境中全局命名服务，服务注册中心，全局分布式锁等等。
之前我们发布一个rpc请求的时候，rpcChannel是直接知道rpc请求在那台服务器上。但实际上并不是的，服务是单独部署的，rpc节点都是对等的，地位都是相同的。想要去使用一个rpc服务时，并不知道rpc在哪台服务器上运行。所以需要一个服务配置中心，所有提供rpc服务的节点都需要向配置中心注册一下一个服务，IP/PORT.

• 文件系统
  Zookeeper提供一个多层级的节点命名空间（节点称为znode）。与文件系统不同的是，这些节点都可以设置关联的数据，而文件系统中只有文件节点可以存放数据而目录节点不行。Zookeeper为了保证高吞吐和低延迟，在内存中维护了这个树状的目录结构，这种特性使得Zookeeper不能用于存放大量的数据，每个节点的存放数据上限为1M。
• 通知机制
  client端会对某个znode建立一个watcher事件，当该znode发生变化时，这些client会收到zk的通知，然后client可以S根据znode变化来做出业务上的改变等。
  

ls、get、create、set、delete

服务的动态注册和发现，为了支持高并发，OrderService被部署了4份，每个客户端都保存了一份服务提供者的列表，但这个列表是静态的（在配置文件中写死的），如果服务的提供者发生了变化，例如有些机器down了，或者又新增了OrderService的实例，客户端根本不知道，想要得到最新的服务提供者的URL列表，必须手工更新配置文件，很不方便。

解除耦合，增加一个中间层 – 注册中心它保存了能提供的服务的名称，以及URL。首先这些服务会在注册中心进行注册，当客户端来查询的时候，只需要给出名称，注册中心就会给出一个URL。所有的客户端在访问服务前，都需要向这个注册中心进行询问，以获得最新的地址。

心跳机制的原理很简单：客户端每隔N秒向服务端发送一个心跳消息，服务端收到心跳消息后，回复同样的心跳消息给客户端。如果服务端或客户端在M秒（M>N）内都没有收到包括心跳消息在内的任何消息，即心跳超时，我们就认为目标TCP连接已经断开了。

RPC节点超时未发送心跳，zk不会删除这个节点

RPC节点超时未发送心跳消息，zk会自动删除临时性节点

zookeeper的watch监听机制，用于监听节点数据变化。当watch监听的数据发生变化时，通知该watch的client，即watcher。

那么既然有watch监听数据变化事件，就有对应的事件类型和状态类型

• 客户端调用一个RPC服务的时候 ，会拿RPC服务的名字以及方法的名字组成一个node节点路径，去服务配置中心查找节点是否存在。如果存在去找ip+port。
• watch机制：监听节点的变化，维护了一个map表，键就是节点名字，值就是节点内容。如果节点变化，zk会主动告诉客户端，这就是一种回调机制。

Zookeeper原生提供了C和Java的客户端编程接口，但是使用起来相对复杂，几个弱点：
1.不会自动发送心跳消息 <==== 错误，源码上会在1/3的Timeout时间发送ping心跳消息
2.设置监听watcher只能是一次性的，每次触发后需要重复设置
3.znode节点只存储简单的byte字节数组，如果存储对象，需要自己转换对象生成字节数组

// 封装的zk客户端类
class ZkClient
{
public:
    ZkClient();
    ~ZkClient();
    // zkclient启动连接zkserver
    void Start();
    // 在zkserver上根据指定的path创建znode节点
    void Create(const char *path, const char *data, int datalen, int state=0);
    // 根据参数指定的znode节点路径，或者znode节点的值
    std::string GetData(const char *path);
private:
    // zk的客户端句柄
    zhandle_t *m_zhandle;
};

#include "zookeeperutil.h"
#include "mprpcapplication.h"
#include 
#include 

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

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

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