RPC框架引入zookeeper服务注册与服务发现-Toy模板网

这篇具有很好参考价值的文章主要介绍了RPC框架引入zookeeper服务注册与服务发现。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

Zookeeper概念及其作用

ZooKeeper是一个分布式的，开放源码的分布式应用程序协调服务，是Google的Chubby一个开源的实现，是大数据生态中的重要组件。它是集群的管理者，监视着集群中各个节点的状态根据节点提交的反馈进行下一步合理操作。最终，将简单易用的接口和性能高效、功能稳定的系统提供给用户。

它是一个为分布式应用提供一致性协调服务的中间件
zookeeper入门参考链接：https://www.cnblogs.com/xinyonghu/p/11031729.html

在分布式系统中，zookeeper提供了非常丰富的应用，本文只是剖析其中一小部分，但也是非常重要的一个部分，即服务注册和服务发现。
在RPC框架中，如果没有服务注册和服务发现，那么这个RPC框架几乎变得不实用，浅显的思路是在RPCConsumer(服务调用端）维护一个服务的列表，这个列表包含了所有分布式节点服务的ip和端口，但考虑这么一种情况，如果其中某个节点由于某种原因down掉了或者将这个节点的服务删除了，但是RPCConsumer本地还维护的列表中还存在这个服务结点，并且还尝试请求这个服务，那么显然会调用出错。
RPC框架引入zookeeper服务注册与服务发现,分布式,rpc,zookeeper,c++,分布式

类似这种肯定需要动态的维护每个分布式服务节点的状态，在该节点down掉或者被撤销时应及时删除这个服务，避免RPC调用端继续请求不存在的服务。这就是zookeeper服务注册和服务发现所做的事。

RPC框架引入zookeeper服务注册与服务发现,分布式,rpc,zookeeper,c++,分布式
Zookeeper组织数据的格式类似于一个文件系统，每个znode结点都可以是一个分布式服务结点，一般组织的结构是XXXXService/login、 XXXXService/registe，即service_name/method_name，znode结点的数据就是该服务所在节点的ip和port
RPC框架引入zookeeper服务注册与服务发现,分布式,rpc,zookeeper,c++,分布式

Zookeeper服务注册和发现的流程：

step1：Rpc服务端先通过zkClient向zkServer端注册服务，也即创建XXXXService/login、 XXXXService/registe节点，并填充相应的数据。
step2：Rpc调用端再调用某个服务之前，通过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 <string>


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

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

        }

3、RPCConsumer端进行服务发现

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