RPC框架引入zookeeper服务注册与服务发现

Zookeeper概念及其作用

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

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

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

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

在这里插入图片描述
Zookeeper组织数据的格式类似于一个文件系统,每个znode结点都可以是一个分布式服务结点,一般组织的结构是XXXXService/login、 XXXXService/registe,即service_name/method_name,znode结点的数据就是该服务所在节点的ip和port
在这里插入图片描述

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启动连接zkservervoid 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   ---> /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);}}

3、RPCConsumer端进行服务发现

//rpc调用方想调用service_name的method_name的服务, 需要查询zk上该服务所在的host信息ZkClient zkCli;zkCli.Start();// /UserServiceRpc/Loginstd::string method_path = "/" + service_name + "/" + method_name;// 127.0.0.1:8000std::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());

至此基本上完整RPC应该具备的核心东西都有了。

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.hqwc.cn/news/55715.html

如若内容造成侵权/违法违规/事实不符,请联系编程知识网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

python-MySQL数据库建表语句(需要连接数据库)转存为Excel文档-工作小记

将create table XXXXXX 转为指定Excel文档。该脚本适用于数据库表结构本地文档记录 呈现效果 代码 # -*- coding:utf-8 -*- # Time : 2023/8/2 15:14 # Author: 水兵没月 # File : MySQL建表_2_excel.py import reimport mysql.connector import pandas as pd db 库名 mydb …

Java自定义校验注解实现List、set集合字段唯一性校验

文章目录 一&#xff1a; 使用场景二&#xff1a; 定义FieldUniqueValid注解2.1 FieldUniqueValid2.2 注解说明2.3 Constraint 注解介绍2.4 FieldUniqueValid注解使用 三&#xff1a;自定义FieldUniqueValidator校验类3.1 实现ConstraintValidator3.2 重写initialize方法3.3 重…

线性规划和单纯形法-原理篇

文章目录 引言线性规划标准型问题特点单纯形法 引言 很多运筹学的教材都是从线性规划开始的&#xff0c;我平时做算法策略的落地应用时也研发了一部分基于线性规划的技术方案。可以说&#xff0c;如果搞不懂线性规划&#xff0c;很难成为一名优秀的运筹优化算法工程师。 但是…

Mybatis实现JsonObject对象与JSON之间交互

项目中使用PostGresql数据库进行数据存储&#xff0c;表中某字段为Json类型&#xff0c;用于存储Json格式数据。PG数据库能够直接存储Json算是一大特色&#xff0c;很多特定情境下使用直接存储Json字段数据能够大量节省开发时间&#xff0c;提高后台数据查询和转换效率。 1、基…

JJWT快速入门

本篇介绍使用 JJWT&#xff08;Java JWT&#xff09;库来生成 JWT Token&#xff0c;步骤如下&#xff1a; 添加依赖&#xff1a; 在项目中添加 JJWT 依赖项。对于 Maven 项目&#xff0c;可以在 pom.xml 文件中添加以下依赖项&#xff1a; <dependency><groupId>…

Linux(进程)

Linux&#xff08;进程&#xff09; 1. 冯诺依曼结构体系2 . 操作系统&#xff08;OS&#xff09;3.进程task_ struct内容分类查看进程查看PID以及PPIDfork()Linux操作系统进程的状态僵尸进程孤儿进程进程优先级其他概念 1. 冯诺依曼结构体系 冯诺依曼结构也称普林斯顿结构&am…

QT生成可执行文件的步骤

QT生成可执行文件的步骤 第一步&#xff1a;debug为release&#xff0c;然后进行编译 第二步&#xff1a;添加QT生成必要的库 首先&#xff0c;建立一个新的文件夹&#xff0c;然后将Release中的可执行文件拷贝到新的文件夹中 然后&#xff0c;在新建文件夹中生成必要的库 …

论文阅读- Uncovering Coordinated Networks on Social Media:Methods and Case Studies

链接&#xff1a;https://arxiv.org/pdf/2001.05658.pdf 目录 摘要&#xff1a; 引言 Methods Case Study 1: Account Handle Sharing Coordination Detection 分析 Case Study 2: Image Coordination Coordination Detection Analysis Case Study 3: Hashtag Sequen…

【C++从0到王者】第十六站:stack和queue的使用

文章目录 一、stack的使用1.stack的介绍2.stack的使用 二、queue的使用1.queue的护额晒2.queue的使用 三、stack和queue相关算法题1.最小栈2.栈的压入、弹出序列3.逆波兰表达式4.两个栈实现一个队列5.用两个队列实现栈6.二叉树的层序遍历1.双队列2.用一个变量levelSize去控制 7…

总结950

7:00起床 7:30~8:00复习单词300个&#xff0c;记忆100个 8:10~9:30数学660&#xff0c;只做了10道题&#xff0c;发现对各知识点的掌握程度不一。有些熟练&#xff0c;有些生疏 9:33~10:25计算机网络课程1h 10:32~12:02继续660&#xff0c;也不知道做了几道 2:32~4:00数据…

SpringBoot第31讲:SpringBoot集成ShardingJDBC - Sharding-JDBC简介和基于MyBatis的单库分表

SpringBoot第31讲&#xff1a;SpringBoot集成ShardingJDBC - Sharding-JDBC简介和基于MyBatis的单库分表 本文是SpringBoot第31讲&#xff0c;主要介绍分表分库&#xff0c;以及SpringBoot集成基于ShardingJDBCMyBatis的单库分表实践 文章目录 SpringBoot第31讲&#xff1a;Spr…

【Python从入门到进阶】31、使用JSONPath解析淘票票网站地区接口数据

接上篇《30、JSONPath的介绍和使用》 上一篇我们介绍了JSONPath的基础和具体使用&#xff0c;本篇我们来具体使用JSONPath&#xff0c;来解析淘票票网站的地区接口数据。 一、引言 1、JsonPath的作用和用途&#xff1f; JsonPath是一种用于在JSON数据中进行查询和提取的表达…